[Bridge] Folded Nexus operation progress onto the next Workflow Task.

chasm/lib/nexusoperation/config.go

@@ -60,6 +60,15 @@ one per operation onto the caller's next Workflow Task scheduled event. When fal
 delivery is refused with a 400, which tells the handler to stop sending progress.`,
 )
 
+var ProgressMinInterval = dynamicconfig.NewNamespaceDurationSetting(
+	"nexusoperation.progressMinInterval",
+	time.Second,
+	`Experimental. The shortest time between a Workflow Task that carried Nexus operation progress
+and the next task that progress schedules for the same caller Workflow. Progress that arrives
+sooner waits, and newer progress folds into it. A task scheduled for any other reason still
+carries it. This bounds the History a long stream costs the caller.`,
+)
+
 var ChasmWorkflowOperationsRolloutPercent = dynamicconfig.NewNamespaceIntSetting(
 	"nexusoperation.chasmWorkflowOperationsRolloutPercent",
 	0,

chasm/lib/nexusoperation/operation.go

@@ -108,6 +108,20 @@ type Operation struct {
 	// Callbacks holds completion callbacks to be invoked when this reaches a terminal state.
 	// Keyed by completionCallbackID(requestID, index).
 	Callbacks chasm.Map[string, *callback.Callback]
+
+	// PendingProgress is the latest progress no Workflow Task has carried yet.
+	PendingProgress chasm.Field[*nexuspb.NexusOperationProgress]
+	// DeliveredProgress is the latest progress a Workflow Task carried. A delivery whose
+	// counter is not above it is stale.
+	DeliveredProgress chasm.Field[*nexuspb.NexusOperationProgress]
+}
+
+// ProgressStore is implemented by a parent that delivers the operation's progress to its caller.
+// A parent without it, such as none for a standalone operation, drops progress.
+type ProgressStore interface {
+	// OnNexusOperationProgress records that the operation holds pending progress. The parent
+	// decides which Workflow Task takes it.
+	OnNexusOperationProgress(ctx chasm.MutableContext, operation *Operation) error
 }
 
 // NewOperation creates a new Operation component with the given persisted state.
@@ -298,10 +312,9 @@ func (o *Operation) HandleNexusCompletion(
 		return serviceerror.NewNotFound("operation not found")
 	}
 
-	// Progress never stands in for the start, and nothing folds it onto a Workflow Task yet, so it
-	// is accepted and dropped here.
-	if completion.GetProgress() != nil {
-		return nil
+	// Progress never stands in for the start, so it is handled before completion-before-start.
+	if progress := completion.GetProgress(); progress != nil {
+		return o.onProgress(ctx, progress)
 	}
 
 	links := completion.GetLinks()
@@ -329,6 +342,56 @@ func (o *Operation) HandleNexusCompletion(
 	}
 }
 
+// onProgress keeps the progress with the highest counter for the caller. Progress before the
+// operation started is dropped, since it cannot stand in for the start response, and so is progress
+// after it closed, since its completion supersedes it.
+func (o *Operation) onProgress(
+	ctx chasm.MutableContext,
+	progress *nexuspb.NexusOperationProgress,
+) error {
+	if o.Status != nexusoperationpb.OPERATION_STATUS_STARTED {
+		return nil
+	}
+	parent, ok := o.Store.TryGet(ctx)
+	if !ok {
+		return nil
+	}
+	store, ok := parent.(ProgressStore)
+	if !ok {
+		return nil
+	}
+	for _, seen := range []chasm.Field[*nexuspb.NexusOperationProgress]{
+		o.PendingProgress,
+		o.DeliveredProgress,
+	} {
+		if previous, ok := seen.TryGet(ctx); ok && previous.GetCounter() >= progress.GetCounter() {
+			return nil
+		}
+	}
+	o.PendingProgress = chasm.NewDataField(ctx, progress)
+	return store.OnNexusOperationProgress(ctx, o)
+}
+
+// TakePendingProgress returns the pending progress and records it as delivered.
+func (o *Operation) TakePendingProgress(
+	ctx chasm.MutableContext,
+) (*nexuspb.NexusOperationProgress, bool) {
+	progress, ok := o.PendingProgress.TryGet(ctx)
+	if !ok {
+		return nil, false
+	}
+	o.DeliveredProgress = chasm.NewDataField(ctx, progress)
+	o.PendingProgress = chasm.NewEmptyField[*nexuspb.NexusOperationProgress]()
+	return progress, true
+}
+
+// DropPendingProgress forgets the pending progress without delivering it.
+func (o *Operation) DropPendingProgress(ctx chasm.MutableContext) {
+	if _, ok := o.PendingProgress.TryGet(ctx); ok {
+		o.PendingProgress = chasm.NewEmptyField[*nexuspb.NexusOperationProgress]()
+	}
+}
+
 // loadStartArgs is a ReadComponent callback that loads the start arguments from the operation.
 func (o *Operation) loadStartArgs(
 	ctx chasm.Context,

chasm/lib/nexusoperation/progress_test.go

@@ -0,0 +1,140 @@
+package nexusoperation
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	nexuspb "go.temporal.io/api/nexus/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/chasm"
+	nexusoperationpb "go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1"
+)
+
+// progressStoreComponent is a parent that takes progress.
+type progressStoreComponent struct {
+	mockStoreComponent
+
+	pending int
+}
+
+func (s *progressStoreComponent) OnNexusOperationProgress(chasm.MutableContext, *Operation) error {
+	s.pending++
+	return nil
+}
+
+func TestHandleNexusProgress(t *testing.T) {
+	progress := func(counter int64) *persistencespb.ChasmNexusCompletion {
+		return &persistencespb.ChasmNexusCompletion{
+			Outcome: &persistencespb.ChasmNexusCompletion_Progress{
+				Progress: &nexuspb.NexusOperationProgress{Counter: counter},
+			},
+		}
+	}
+	newOp := func(status nexusoperationpb.OperationStatus, store OperationStore) *Operation {
+		op := newTestOperation()
+		op.Status = status
+		if store != nil {
+			op.Store = chasm.NewMockParentPtr(store)
+		}
+		return op
+	}
+	pendingCounter := func(ctx chasm.Context, op *Operation) int64 {
+		p, _ := op.PendingProgress.TryGet(ctx)
+		return p.GetCounter()
+	}
+	deliveredCounter := func(ctx chasm.Context, op *Operation) int64 {
+		p, _ := op.DeliveredProgress.TryGet(ctx)
+		return p.GetCounter()
+	}
+
+	for _, tc := range []struct {
+		name        string
+		status      nexusoperationpb.OperationStatus
+		store       OperationStore
+		counters    []int64
+		wantPending int64
+	}{
+		{
+			name:        "HighestCounterWins",
+			status:      nexusoperationpb.OPERATION_STATUS_STARTED,
+			store:       &progressStoreComponent{},
+			counters:    []int64{1, 3, 2},
+			wantPending: 3,
+		},
+		{
+			name:     "DroppedBeforeStart",
+			status:   nexusoperationpb.OPERATION_STATUS_SCHEDULED,
+			store:    &progressStoreComponent{},
+			counters: []int64{1},
+		},
+		{
+			name:     "DroppedAfterClose",
+			status:   nexusoperationpb.OPERATION_STATUS_SUCCEEDED,
+			store:    &progressStoreComponent{},
+			counters: []int64{1},
+		},
+		{
+			name:     "DroppedWithoutCaller",
+			status:   nexusoperationpb.OPERATION_STATUS_STARTED,
+			counters: []int64{1},
+		},
+		{
+			name:     "DroppedByAParentThatTakesNoProgress",
+			status:   nexusoperationpb.OPERATION_STATUS_STARTED,
+			store:    &mockStoreComponent{},
+			counters: []int64{1},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := &chasm.MockMutableContext{}
+			op := newOp(tc.status, tc.store)
+			for _, counter := range tc.counters {
+				require.NoError(t, op.HandleNexusCompletion(ctx, progress(counter)))
+			}
+			require.Equal(t, tc.wantPending, pendingCounter(ctx, op))
+			require.Equal(t, tc.status, op.Status, "progress never changes the operation status")
+		})
+	}
+
+	t.Run("StaleAfterDelivery", func(t *testing.T) {
+		ctx := &chasm.MockMutableContext{}
+		store := &progressStoreComponent{}
+		op := newOp(nexusoperationpb.OPERATION_STATUS_STARTED, store)
+		require.NoError(t, op.HandleNexusCompletion(ctx, progress(5)))
+		taken, ok := op.TakePendingProgress(ctx)
+		require.True(t, ok)
+		require.Equal(t, int64(5), taken.GetCounter())
+		require.Zero(t, pendingCounter(ctx, op))
+
+		require.NoError(t, op.HandleNexusCompletion(ctx, progress(4)))
+		require.Zero(t, pendingCounter(ctx, op), "a counter at or below the delivered one is stale")
+		require.NoError(t, op.HandleNexusCompletion(ctx, progress(6)))
+		require.Equal(t, int64(6), pendingCounter(ctx, op))
+		require.Equal(t, 2, store.pending)
+	})
+
+	t.Run("NewerProgressWaitsUntilATaskCarriesIt", func(t *testing.T) {
+		ctx := &chasm.MockMutableContext{}
+		store := &progressStoreComponent{}
+		op := newOp(nexusoperationpb.OPERATION_STATUS_STARTED, store)
+		require.NoError(t, op.HandleNexusCompletion(ctx, progress(1)))
+		_, ok := op.TakePendingProgress(ctx)
+		require.True(t, ok)
+
+		// Whether it folds into a task that has not started is the parent's call; the operation
+		// keeps it until a task carries it.
+		require.NoError(t, op.HandleNexusCompletion(ctx, progress(2)))
+		require.Equal(t, int64(2), pendingCounter(ctx, op))
+		require.Equal(t, int64(1), deliveredCounter(ctx, op))
+		require.Equal(t, 2, store.pending)
+	})
+
+	t.Run("DropPendingProgress", func(t *testing.T) {
+		ctx := &chasm.MockMutableContext{}
+		op := newOp(nexusoperationpb.OPERATION_STATUS_STARTED, &progressStoreComponent{})
+		require.NoError(t, op.HandleNexusCompletion(ctx, progress(1)))
+		op.DropPendingProgress(ctx)
+		require.Zero(t, pendingCounter(ctx, op))
+		require.Zero(t, deliveredCounter(ctx, op), "dropped progress is not delivered")
+	})
+}