Add CloseTime to OperationCompletion* structs (#59)

What?
--
- Added `CloseTime` to the various OperationCompletion* structs.

Why?
--
- `CloseTime` is useful in determining an order of events from
completion callbacks. We already have `StartTime`, but adding
`CloseTime` adds flexibility and is generally useful as for metrics.
- In CHASM Scheduler, `CloseTime` is used to determine the earliest
possible time our system could kick off a buffered start in sequence of
a prior start completing (drives a customer-facing metric).
- If we weren't to add this field in Nexus, we'd have to make an
additional follow-up `Describe` call to determine it, as part of
processing the Nexus callback.

---------

Co-authored-by: Roey Berman <roey.berman@gmail.com>

Author: lina-temporal

nexus/api.go

@@ -26,6 +26,7 @@ const (
 	headerRequestID          = "nexus-request-id"
 	headerLink               = "nexus-link"
 	headerOperationStartTime = "nexus-operation-start-time"
+	headerOperationCloseTime = "nexus-operation-close-time"
 	headerRetryable          = "nexus-request-retryable"
 	// HeaderOperationID is the unique ID returned by the StartOperation response for async operations.
 	// Must be set on callback headers to support completing operations before the start response is received.

nexus/completion.go

@@ -57,6 +57,8 @@ type OperationCompletionSuccessful struct {
 	OperationToken string
 	// StartTime is the time the operation started. Used when a completion callback is received before a started response.
 	StartTime time.Time
+	// CloseTime is the time the operation completed. Used when a completion callback is received before a started response.
+	CloseTime time.Time
 	// Links are used to link back to the operation when a completion callback is received before a started response.
 	Links []Link
 }
@@ -77,6 +79,8 @@ type OperationCompletionSuccessfulOptions struct {
 	OperationToken string
 	// StartTime is the time the operation started. Used when a completion callback is received before a started response.
 	StartTime time.Time
+	// CloseTime is the time the operation completed. Used when a completion callback is received before a started response.
+	CloseTime time.Time
 	// Links are used to link back to the operation when a completion callback is received before a started response.
 	Links []Link
 }
@@ -117,6 +121,7 @@ func NewOperationCompletionSuccessful(result any, options OperationCompletionSuc
 		OperationID:    options.OperationID,
 		OperationToken: options.OperationToken,
 		StartTime:      options.StartTime,
+		CloseTime:      options.CloseTime,
 		Links:          options.Links,
 	}, nil
 }
@@ -147,6 +152,9 @@ func (c *OperationCompletionSuccessful) applyToHTTPRequest(request *http.Request
 	if c.Header.Get(headerOperationStartTime) == "" && !c.StartTime.IsZero() {
 		request.Header.Set(headerOperationStartTime, c.StartTime.Format(http.TimeFormat))
 	}
+	if c.Header.Get(headerOperationCloseTime) == "" && !c.CloseTime.IsZero() {
+		request.Header.Set(headerOperationCloseTime, c.CloseTime.Format(time.RFC3339Nano))
+	}
 	if c.Header.Get(headerLink) == "" {
 		if err := addLinksToHTTPHeader(c.Links, request.Header); err != nil {
 			return err
@@ -176,6 +184,8 @@ type OperationCompletionUnsuccessful struct {
 	OperationToken string
 	// StartTime is the time the operation started. Used when a completion callback is received before a started response.
 	StartTime time.Time
+	// CloseTime is the time the operation completed. This may be different from the time the completion callback is delivered.
+	CloseTime time.Time
 	// Links are used to link back to the operation when a completion callback is received before a started response.
 	Links []Link
 	// Failure object to send with the completion.
@@ -198,6 +208,8 @@ type OperationCompletionUnsuccessfulOptions struct {
 	OperationToken string
 	// StartTime is the time the operation started. Used when a completion callback is received before a started response.
 	StartTime time.Time
+	// CloseTime is the time the operation completed. This may be different from the time the completion callback is delivered.
+	CloseTime time.Time
 	// Links are used to link back to the operation when a completion callback is received before a started response.
 	Links []Link
 }
@@ -217,6 +229,7 @@ func NewOperationCompletionUnsuccessful(error *OperationError, options Operation
 		OperationID:    options.OperationID,
 		OperationToken: options.OperationToken,
 		StartTime:      options.StartTime,
+		CloseTime:      options.CloseTime,
 		Links:          options.Links,
 	}, nil
 }
@@ -245,6 +258,9 @@ func (c *OperationCompletionUnsuccessful) applyToHTTPRequest(request *http.Reque
 	if c.Header.Get(headerOperationStartTime) == "" && !c.StartTime.IsZero() {
 		request.Header.Set(headerOperationStartTime, c.StartTime.Format(http.TimeFormat))
 	}
+	if c.Header.Get(headerOperationCloseTime) == "" && !c.CloseTime.IsZero() {
+		request.Header.Set(headerOperationCloseTime, c.CloseTime.Format(time.RFC3339Nano))
+	}
 	if c.Header.Get(headerLink) == "" {
 		if err := addLinksToHTTPHeader(c.Links, request.Header); err != nil {
 			return err
@@ -277,6 +293,8 @@ type CompletionRequest struct {
 	OperationToken string
 	// StartTime is the time the operation started. Used when a completion callback is received before a started response.
 	StartTime time.Time
+	// CloseTime is the time the operation completed. This may be different from the time the completion callback is delivered.
+	CloseTime time.Time
 	// Links are used to link back to the operation when a completion callback is received before a started response.
 	Links []Link
 	// Parsed from request and set if State is failed or canceled.
@@ -335,6 +353,13 @@ func (h *completionHTTPHandler) ServeHTTP(writer http.ResponseWriter, request *h
 			return
 		}
 	}
+	if closeTimeHeader := request.Header.Get(headerOperationCloseTime); closeTimeHeader != "" {
+		var parseTimeErr error
+		if completion.CloseTime, parseTimeErr = time.Parse(time.RFC3339Nano, closeTimeHeader); parseTimeErr != nil {
+			h.writeFailure(writer, HandlerErrorf(HandlerErrorTypeBadRequest, "failed to parse operation close time header"))
+			return
+		}
+	}
 	var decodeErr error
 	if completion.Links, decodeErr = getLinksFromHeader(request.Header); decodeErr != nil {
 		h.writeFailure(writer, HandlerErrorf(HandlerErrorTypeBadRequest, "failed to decode links from request headers"))

nexus/completion_test.go

@@ -13,6 +13,20 @@ import (
 )
 
 type successfulCompletionHandler struct {
+	expectedStartTime time.Time
+	expectedCloseTime time.Time
+}
+
+// validateExpectedTime returns false if the times are set but aren't equal.
+func validateExpectedTime(expected, actual time.Time, resolution time.Duration) bool {
+	if expected.IsZero() {
+		return true
+	}
+
+	expected = expected.Truncate(resolution)
+	actual = actual.Truncate(resolution)
+
+	return expected.Equal(actual)
 }
 
 func (h *successfulCompletionHandler) CompleteOperation(ctx context.Context, completion *CompletionRequest) error {
@@ -37,6 +51,12 @@ func (h *successfulCompletionHandler) CompleteOperation(ctx context.Context, com
 	if len(completion.Links) == 0 {
 		return HandlerErrorf(HandlerErrorTypeBadRequest, "expected Links to be set on CompletionRequest")
 	}
+	if !validateExpectedTime(h.expectedStartTime, completion.StartTime, time.Second) {
+		return HandlerErrorf(HandlerErrorTypeBadRequest, "expected StartTime to be equal")
+	}
+	if !validateExpectedTime(h.expectedCloseTime, completion.CloseTime, time.Millisecond) {
+		return HandlerErrorf(HandlerErrorTypeBadRequest, "expected CloseTime to be equal")
+	}
 	var result int
 	err := completion.Result.Consume(&result)
 	if err != nil {
@@ -49,12 +69,19 @@ func (h *successfulCompletionHandler) CompleteOperation(ctx context.Context, com
 }
 
 func TestSuccessfulCompletion(t *testing.T) {
-	ctx, callbackURL, teardown := setupForCompletion(t, &successfulCompletionHandler{}, nil, nil)
+	startTime := time.Now().Add(-time.Hour).UTC()
+	closeTime := time.Now().UTC()
+
+	ctx, callbackURL, teardown := setupForCompletion(t, &successfulCompletionHandler{
+		expectedStartTime: startTime,
+		expectedCloseTime: closeTime,
+	}, nil, nil)
 	defer teardown()
 
 	completion, err := NewOperationCompletionSuccessful(666, OperationCompletionSuccessfulOptions{
 		OperationToken: "test-operation-token",
-		StartTime:      time.Now(),
+		StartTime:      startTime,
+		CloseTime:      closeTime,
 		Links: []Link{{
 			URL: &url.URL{
 				Scheme:   "https",
@@ -113,7 +140,9 @@ func TestSuccessfulCompletion_CustomSerializer(t *testing.T) {
 }
 
 type failureExpectingCompletionHandler struct {
-	errorChecker func(error) error
+	errorChecker      func(error) error
+	expectedStartTime time.Time
+	expectedCloseTime time.Time
 }
 
 func (h *failureExpectingCompletionHandler) CompleteOperation(ctx context.Context, completion *CompletionRequest) error {
@@ -135,24 +164,36 @@ func (h *failureExpectingCompletionHandler) CompleteOperation(ctx context.Contex
 	if len(completion.Links) == 0 {
 		return HandlerErrorf(HandlerErrorTypeBadRequest, "expected Links to be set on CompletionRequest")
 	}
+	if !validateExpectedTime(h.expectedStartTime, completion.StartTime, time.Second) {
+		return HandlerErrorf(HandlerErrorTypeBadRequest, "expected StartTime to be equal")
+	}
+	if !validateExpectedTime(h.expectedCloseTime, completion.CloseTime, time.Millisecond) {
+		return HandlerErrorf(HandlerErrorTypeBadRequest, "expected CloseTime to be equal")
+	}
 
 	return nil
 }
 
 func TestFailureCompletion(t *testing.T) {
+	startTime := time.Now().Add(-time.Hour).UTC()
+	closeTime := time.Now().UTC()
+
 	ctx, callbackURL, teardown := setupForCompletion(t, &failureExpectingCompletionHandler{
 		errorChecker: func(err error) error {
 			if err.Error() != "expected message" {
 				return HandlerErrorf(HandlerErrorTypeBadRequest, "invalid failure: %v", err)
 			}
 			return nil
 		},
+		expectedStartTime: startTime,
+		expectedCloseTime: closeTime,
 	}, nil, nil)
 	defer teardown()
 
 	completion, err := NewOperationCompletionUnsuccessful(NewCanceledOperationError(errors.New("expected message")), OperationCompletionUnsuccessfulOptions{
 		OperationToken: "test-operation-token",
-		StartTime:      time.Now(),
+		StartTime:      startTime,
+		CloseTime:      closeTime,
 		Links: []Link{{
 			URL: &url.URL{
 				Scheme:   "https",
@@ -177,20 +218,26 @@ func TestFailureCompletion(t *testing.T) {
 
 func TestFailureCompletion_CustomFailureConverter(t *testing.T) {
 	fc := customFailureConverter{}
+	startTime := time.Now().Add(-time.Hour).UTC()
+	closeTime := time.Now().UTC()
+
 	ctx, callbackURL, teardown := setupForCompletion(t, &failureExpectingCompletionHandler{
 		errorChecker: func(err error) error {
 			if !errors.Is(err, errCustom) {
 				return HandlerErrorf(HandlerErrorTypeBadRequest, "invalid failure, expected a custom error: %v", err)
 			}
 			return nil
 		},
+		expectedStartTime: startTime,
+		expectedCloseTime: closeTime,
 	}, nil, fc)
 	defer teardown()
 
 	completion, err := NewOperationCompletionUnsuccessful(NewCanceledOperationError(errors.New("expected message")), OperationCompletionUnsuccessfulOptions{
 		FailureConverter: fc,
 		OperationToken:   "test-operation-token",
-		StartTime:        time.Now(),
+		StartTime:        startTime,
+		CloseTime:        closeTime,
 		Links: []Link{{
 			URL: &url.URL{
 				Scheme:   "https",