CHASM update consistency level

Author: stephanos

chasm/engine.go

@@ -24,27 +24,27 @@ type Engine interface {
 		context.Context,
 		ComponentRef,
 		func(MutableContext) (RootComponent, error),
-		func(MutableContext, Component) error,
+		func(MutableContext, Component, *Registry) error,
 		...TransitionOption,
 	) (EngineUpdateWithStartExecutionResult, error)
 
 	UpdateComponent(
 		context.Context,
 		ComponentRef,
-		func(MutableContext, Component) error,
+		func(MutableContext, Component, *Registry) error,
 		...TransitionOption,
 	) ([]byte, error)
 	ReadComponent(
 		context.Context,
 		ComponentRef,
-		func(Context, Component) error,
+		func(Context, Component, *Registry) error,
 		...TransitionOption,
 	) error
 
 	PollComponent(
 		context.Context,
 		ComponentRef,
-		func(Context, Component) (bool, error),
+		func(Context, Component, *Registry) (bool, error),
 		...TransitionOption,
 	) ([]byte, error)
 
@@ -65,6 +65,25 @@ type DeleteExecutionRequest struct {
 	TerminateComponentRequest
 }
 
+// ConsistencyLevel controls how strictly a [ComponentRef] is validated
+// against the current execution state. Ordered from strongest to loosest.
+type ConsistencyLevel int
+
+const (
+	// ConsistencyLevelExecution is the default. Validates [ComponentRef.executionLastUpdateVT].
+	ConsistencyLevelExecution ConsistencyLevel = iota
+	// ConsistencyLevelComponent validates only [ComponentRef.componentInitialVT].
+	// Use when the token may have been generated before a failover changed the
+	// execution-level VT, but the component itself is still valid on the current branch.
+	ConsistencyLevelComponent
+	// ConsistencyLevelBusinessID skips all [VersionedTransition] checks and matches
+	// by [ComponentRef.componentPath] only. Falls back to the latest open run if the
+	// referenced [ComponentRef.RunID] points to a closed execution.
+	// Use when the original run may have been reset: the operation should target the
+	// same component in the new run, matched by path and deduplicated by request ID.
+	ConsistencyLevelBusinessID
+)
+
 type BusinessIDReusePolicy int
 
 const (
@@ -82,10 +101,11 @@ const (
 )
 
 type TransitionOptions struct {
-	ReusePolicy    BusinessIDReusePolicy
-	ConflictPolicy BusinessIDConflictPolicy
-	RequestID      string
-	Speculative    bool
+	ReusePolicy      BusinessIDReusePolicy
+	ConflictPolicy   BusinessIDConflictPolicy
+	RequestID        string
+	Speculative      bool
+	ConsistencyLevel ConsistencyLevel
 }
 
 type TransitionOption func(*TransitionOptions)
@@ -172,6 +192,16 @@ func WithRequestID(
 	}
 }
 
+// WithConsistencyLevel sets the consistency level used when resolving a component reference.
+// See [ConsistencyLevel] for details on each level.
+func WithConsistencyLevel(
+	level ConsistencyLevel,
+) TransitionOption {
+	return func(opts *TransitionOptions) {
+		opts.ConsistencyLevel = level
+	}
+}
+
 // Not needed for V1
 // func WithEagerLoading(
 // 	paths []ComponentPath,
@@ -255,7 +285,7 @@ func UpdateWithStartExecution[C RootComponent, I any, O any](
 			c, err = startFn(mutableContext, input)
 			return c, err
 		},
-		func(mutableContext MutableContext, c Component) (retErr error) {
+		func(mutableContext MutableContext, c Component, _ *Registry) (retErr error) {
 			defer log.CapturePanic(mutableContext.Logger(), &retErr)
 
 			var err error
@@ -306,7 +336,7 @@ func UpdateComponent[C any, R []byte | ComponentRef, I any, O any](
 	newSerializedRef, err := engineFromContext(ctx).UpdateComponent(
 		ctx,
 		ref,
-		func(mutableContext MutableContext, c Component) (retErr error) {
+		func(mutableContext MutableContext, c Component, _ *Registry) (retErr error) {
 			defer log.CapturePanic(mutableContext.Logger(), &retErr)
 
 			var err error
@@ -345,7 +375,7 @@ func ReadComponent[C any, R []byte | ComponentRef, I any, O any](
 	err = engineFromContext(ctx).ReadComponent(
 		ctx,
 		ref,
-		func(chasmContext Context, c Component) (retErr error) {
+		func(chasmContext Context, c Component, _ *Registry) (retErr error) {
 			defer log.CapturePanic(chasmContext.Logger(), &retErr)
 
 			var err error
@@ -386,7 +416,7 @@ func PollComponent[C any, R []byte | ComponentRef, I any, O any](
 	newSerializedRef, err := engineFromContext(ctx).PollComponent(
 		ctx,
 		ref,
-		func(chasmContext Context, c Component) (_ bool, retErr error) {
+		func(chasmContext Context, c Component, _ *Registry) (_ bool, retErr error) {
 			defer log.CapturePanic(chasmContext.Logger(), &retErr)
 
 			out, satisfied, err := monotonicPredicate(

chasm/engine_mock.go

@@ -119,7 +119,7 @@ func (s *fieldSuite) TestFieldGetComponent() {
 
 	chasmContext := NewMutableContext(context.Background(), node)
 
-	c, err := node.Component(chasmContext, ComponentRef{componentPath: rootPath})
+	c, err := node.Component(chasmContext, ComponentRef{componentPath: rootPath}, ConsistencyLevelExecution)
 	s.NoError(err)
 	s.NotNil(c)
 
@@ -216,7 +216,7 @@ func (s *fieldSuite) TestDeferredPointerResolution() {
 
 	// Get components from tree to mark nodes as needing sync.
 
-	rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{})
+	rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{}, ConsistencyLevelExecution)
 	s.NoError(err)
 	rootComponent = rootComponentInterface.(*TestComponent)
 	sc1 = rootComponent.SubComponent1.Get(ctx)
@@ -293,7 +293,7 @@ func (s *fieldSuite) TestMixedPointerScenario() {
 	s.NoError(err)
 
 	// Get components from tree to mark nodes as needing sync.
-	rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{})
+	rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{}, ConsistencyLevelExecution)
 	s.NoError(err)
 	rootComponent = rootComponentInterface.(*TestComponent)
 
@@ -308,7 +308,7 @@ func (s *fieldSuite) TestMixedPointerScenario() {
 	// otherwise those nodes will not be marked as dirty.
 
 	ctx2 := NewMutableContext(context.Background(), rootNode)
-	rootComponentInterface, err = rootNode.Component(ctx2, ComponentRef{})
+	rootComponentInterface, err = rootNode.Component(ctx2, ComponentRef{}, ConsistencyLevelExecution)
 	s.NoError(err)
 
 	rootComponent = rootComponentInterface.(*TestComponent)
@@ -372,7 +372,7 @@ func (s *fieldSuite) TestUnresolvableDeferredPointerError() {
 	s.NoError(err)
 
 	// Get component from tree to mark node as needing sync.
-	rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{})
+	rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{}, ConsistencyLevelExecution)
 	s.NoError(err)
 	rootComponent = rootComponentInterface.(*TestComponent)
 

chasm/field_test.go

@@ -229,7 +229,7 @@ func TestExecuteInvocationTaskNexus_Outcomes(t *testing.T) {
 				gomock.Any(),
 				gomock.Any(),
 				gomock.Any(),
-			).DoAndReturn(func(ctx context.Context, ref chasm.ComponentRef, readFn func(chasm.Context, chasm.Component) error, opts ...chasm.TransitionOption) error {
+			).DoAndReturn(func(ctx context.Context, ref chasm.ComponentRef, readFn func(chasm.Context, chasm.Component, *chasm.Registry) error, opts ...chasm.TransitionOption) error {
 				mockCtx := &chasm.MockContext{
 					HandleNow: func(component chasm.Component) time.Time {
 						return timeSource.Now()
@@ -238,14 +238,14 @@ func TestExecuteInvocationTaskNexus_Outcomes(t *testing.T) {
 						return []byte{}, nil
 					},
 				}
-				return readFn(mockCtx, callback)
+				return readFn(mockCtx, callback, nil)
 			})
 
 			mockEngine.EXPECT().UpdateComponent(
 				gomock.Any(),
 				gomock.Any(),
 				gomock.Any(),
-			).DoAndReturn(func(ctx context.Context, ref chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component) error, opts ...chasm.TransitionOption) ([]any, error) {
+			).DoAndReturn(func(ctx context.Context, ref chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component, *chasm.Registry) error, opts ...chasm.TransitionOption) ([]any, error) {
 				mockCtx := &chasm.MockMutableContext{
 					MockContext: chasm.MockContext{
 						HandleNow: func(component chasm.Component) time.Time {
@@ -256,7 +256,7 @@ func TestExecuteInvocationTaskNexus_Outcomes(t *testing.T) {
 						},
 					},
 				}
-				err := updateFn(mockCtx, callback)
+				err := updateFn(mockCtx, callback, nil)
 				return nil, err
 			})
 
@@ -613,7 +613,7 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 				gomock.Any(),
 				gomock.Any(),
 				gomock.Any(),
-			).DoAndReturn(func(ctx context.Context, ref chasm.ComponentRef, readFn func(chasm.Context, chasm.Component) error, opts ...chasm.TransitionOption) error {
+			).DoAndReturn(func(ctx context.Context, ref chasm.ComponentRef, readFn func(chasm.Context, chasm.Component, *chasm.Registry) error, opts ...chasm.TransitionOption) error {
 				// Create a mock context
 				mockCtx := &chasm.MockContext{
 					HandleNow: func(component chasm.Component) time.Time {
@@ -632,14 +632,14 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 				}
 
 				// Call the readFn with our callback
-				return readFn(mockCtx, callback)
+				return readFn(mockCtx, callback, nil)
 			})
 
 			mockEngine.EXPECT().UpdateComponent(
 				gomock.Any(),
 				gomock.Any(),
 				gomock.Any(),
-			).DoAndReturn(func(ctx context.Context, ref chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component) error, opts ...chasm.TransitionOption) ([]any, error) {
+			).DoAndReturn(func(ctx context.Context, ref chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component, *chasm.Registry) error, opts ...chasm.TransitionOption) ([]any, error) {
 				// Create a mock mutable context
 				mockCtx := &chasm.MockMutableContext{
 					MockContext: chasm.MockContext{
@@ -653,7 +653,7 @@ func TestExecuteInvocationTaskChasm_Outcomes(t *testing.T) {
 				}
 
 				// Call the updateFn with our callback
-				err := updateFn(mockCtx, callback)
+				err := updateFn(mockCtx, callback, nil)
 				return nil, err
 			})
 

chasm/lib/callback/tasks_test.go

@@ -29,7 +29,7 @@ type backfillTestCase struct {
 
 func runBackfillTestCase(t *testing.T, env *testEnv, c *backfillTestCase) {
 	ctx := env.MutableContext()
-	schedComponent, err := env.Node.Component(ctx, chasm.ComponentRef{})
+	schedComponent, err := env.Node.Component(ctx, chasm.ComponentRef{}, chasm.ConsistencyLevelExecution)
 	require.NoError(t, err)
 	sched := schedComponent.(*scheduler.Scheduler)
 	invoker := sched.Invoker.Get(ctx)
@@ -229,7 +229,7 @@ func TestBackfillTask_PartialFill(t *testing.T) {
 	}
 
 	ctx := env.MutableContext()
-	schedComponent, err := env.Node.Component(ctx, chasm.ComponentRef{})
+	schedComponent, err := env.Node.Component(ctx, chasm.ComponentRef{}, chasm.ConsistencyLevelExecution)
 	require.NoError(t, err)
 	sched := schedComponent.(*scheduler.Scheduler)
 	backfiller := sched.NewRangeBackfiller(ctx, request)
@@ -241,7 +241,7 @@ func TestBackfillTask_PartialFill(t *testing.T) {
 
 	// Backfiller should still exist (not complete).
 	ctx = env.MutableContext()
-	schedComponent, err = env.Node.Component(ctx, chasm.ComponentRef{})
+	schedComponent, err = env.Node.Component(ctx, chasm.ComponentRef{}, chasm.ConsistencyLevelExecution)
 	require.NoError(t, err)
 	sched = schedComponent.(*scheduler.Scheduler)
 	_, ok := sched.Backfillers[backfiller.BackfillId].TryGet(ctx)

chasm/lib/scheduler/backfiller_tasks_test.go

@@ -299,8 +299,8 @@ func (e *testEnv) ExpectReadComponent(ctx chasm.Context, returnedComponent chasm
 		e.t.Fatal("ExpectReadComponent requires withMockEngine() option")
 	}
 	e.MockEngine.EXPECT().ReadComponent(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
-		DoAndReturn(func(_ context.Context, _ chasm.ComponentRef, readFn func(chasm.Context, chasm.Component) error, _ ...chasm.TransitionOption) error {
-			return readFn(ctx, returnedComponent)
+		DoAndReturn(func(_ context.Context, _ chasm.ComponentRef, readFn func(chasm.Context, chasm.Component, *chasm.Registry) error, _ ...chasm.TransitionOption) error {
+			return readFn(ctx, returnedComponent, nil)
 		}).Times(1)
 }
 
@@ -310,8 +310,8 @@ func (e *testEnv) ExpectUpdateComponent(ctx chasm.MutableContext, componentToUpd
 		e.t.Fatal("ExpectUpdateComponent requires withMockEngine() option")
 	}
 	e.MockEngine.EXPECT().UpdateComponent(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
-		DoAndReturn(func(_ context.Context, _ chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component) error, _ ...chasm.TransitionOption) ([]byte, error) {
-			err := updateFn(ctx, componentToUpdate)
+		DoAndReturn(func(_ context.Context, _ chasm.ComponentRef, updateFn func(chasm.MutableContext, chasm.Component, *chasm.Registry) error, _ ...chasm.TransitionOption) ([]byte, error) {
+			err := updateFn(ctx, componentToUpdate, nil)
 			return nil, err
 		}).Times(1)
 }

chasm/lib/scheduler/helper_test.go

@@ -54,6 +54,18 @@ type ComponentRef struct {
 	validationFn func(NodeBackend, Context, Component, *Registry) error
 }
 
+// RelaxToBusinessID prepares a ComponentRef for cross-run resolution by clearing all
+// versioned transition fields. After this call, the ref identifies a component solely
+// by its path within the execution tree. The RunID is also cleared so that the framework
+// will resolve the ref against the latest open execution for the business ID.
+// This is used internally when ConsistencyLevelBusinessID is in effect and the original
+// run is closed.
+func (r *ComponentRef) RelaxToBusinessID() {
+	r.RunID = ""
+	r.executionLastUpdateVT = nil
+	r.componentInitialVT = nil
+}
+
 // NewComponentRef creates a new ComponentRef with a registered root component go type.
 //
 // In V1, if you don't have a ref,

chasm/ref.go

@@ -98,3 +98,40 @@ func (s *componentRefSuite) TestSerializeDeserialize() {
 	s.Equal(ref.ExecutionKey, deserializedRef.ExecutionKey)
 	s.Equal(ref.componentPath, deserializedRef.componentPath)
 }
+
+func (s *componentRefSuite) TestRelaxToBusinessID() {
+	ref := ComponentRef{
+		ExecutionKey: ExecutionKey{
+			NamespaceID: primitives.NewUUID().String(),
+			BusinessID:  primitives.NewUUID().String(),
+			RunID:       primitives.NewUUID().String(),
+		},
+		archetypeID: 42,
+		executionLastUpdateVT: &persistencespb.VersionedTransition{
+			NamespaceFailoverVersion: rand.Int63(),
+			TransitionCount:          rand.Int63(),
+		},
+		componentPath: []string{"root", "child"},
+		componentInitialVT: &persistencespb.VersionedTransition{
+			NamespaceFailoverVersion: rand.Int63(),
+			TransitionCount:          rand.Int63(),
+		},
+	}
+
+	originalNamespaceID := ref.NamespaceID
+	originalBusinessID := ref.BusinessID
+	originalComponentPath := ref.componentPath
+
+	ref.RelaxToBusinessID()
+
+	// RunID and all VTs should be cleared.
+	s.Empty(ref.RunID)
+	s.Nil(ref.executionLastUpdateVT)
+	s.Nil(ref.componentInitialVT)
+
+	// Other fields should be preserved.
+	s.Equal(originalNamespaceID, ref.NamespaceID)
+	s.Equal(originalBusinessID, ref.BusinessID)
+	s.Equal(ArchetypeID(42), ref.archetypeID)
+	s.Equal(originalComponentPath, ref.componentPath)
+}