From 6258c4aab887fe1674adb7137fe6666c656e0b41 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Tue, 2 Jun 2026 12:50:17 -0700 Subject: [PATCH] chasm: persist links and user_metadata at framework level (#10409) ## What changed? Add per-request links (keyed by request_id) and user_metadata fields to ChasmComponentAttributes so any CHASM component can store them uniformly without each library defining its own proto field. This first PR just adds support, I will open up follow up PRs to update any existing components use these new fields. ## Why? Per the discussion [here](https://github.com/temporalio/temporal/pull/10368#discussion_r3312238761) we think links and user metadata will be common components across CHASM components so it makes sense for the framework to handle this for us vs each component implementing it themselves. ## How did you test it? - [x] built - [ ] run locally and tested manually - [] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) --- api/persistence/v1/chasm.pb.go | 179 ++++++--- chasm/context.go | 36 ++ chasm/context_mock.go | 64 +++- chasm/tree.go | 182 ++++++++- chasm/tree_test.go | 359 ++++++++++++++++++ .../server/api/persistence/v1/chasm.proto | 14 + 6 files changed, 783 insertions(+), 51 deletions(-) diff --git a/api/persistence/v1/chasm.pb.go b/api/persistence/v1/chasm.pb.go index ee1a217adf..0b40344dd5 100644 --- a/api/persistence/v1/chasm.pb.go +++ b/api/persistence/v1/chasm.pb.go @@ -12,7 +12,8 @@ import ( unsafe "unsafe" v1 "go.temporal.io/api/common/v1" - v11 "go.temporal.io/api/failure/v1" + v12 "go.temporal.io/api/failure/v1" + v11 "go.temporal.io/api/sdk/v1" protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" timestamppb "google.golang.org/protobuf/types/known/timestamppb" @@ -225,7 +226,12 @@ type ChasmComponentAttributes struct { // When true, this component ignores parent lifecycle validation. // Detached components can continue operating, accepting writes and executing // tasks, even when their parent is closed/terminated. - Detached bool `protobuf:"varint,4,opt,name=detached,proto3" json:"detached,omitempty"` + Detached bool `protobuf:"varint,4,opt,name=detached,proto3" json:"detached,omitempty"` + // Per-request metadata contributed to this component, keyed by the caller's + // request_id. + Requests map[string]*ChasmComponentAttributes_RequestMetadata `protobuf:"bytes,5,rep,name=requests,proto3" json:"requests,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // Caller-supplied user metadata (summary, details) attached to this component. + UserMetadata *v11.UserMetadata `protobuf:"bytes,6,opt,name=user_metadata,json=userMetadata,proto3" json:"user_metadata,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -288,6 +294,20 @@ func (x *ChasmComponentAttributes) GetDetached() bool { return false } +func (x *ChasmComponentAttributes) GetRequests() map[string]*ChasmComponentAttributes_RequestMetadata { + if x != nil { + return x.Requests + } + return nil +} + +func (x *ChasmComponentAttributes) GetUserMetadata() *v11.UserMetadata { + if x != nil { + return x.UserMetadata + } + return nil +} + type ChasmDataAttributes struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -686,7 +706,7 @@ func (x *ChasmNexusCompletion) GetSuccess() *v1.Payload { return nil } -func (x *ChasmNexusCompletion) GetFailure() *v11.Failure { +func (x *ChasmNexusCompletion) GetFailure() *v12.Failure { if x != nil { if x, ok := x.Outcome.(*ChasmNexusCompletion_Failure); ok { return x.Failure @@ -741,7 +761,7 @@ type ChasmNexusCompletion_Success struct { type ChasmNexusCompletion_Failure struct { // Operation failure, only set if state != successful. - Failure *v11.Failure `protobuf:"bytes,2,opt,name=failure,proto3,oneof"` + Failure *v12.Failure `protobuf:"bytes,2,opt,name=failure,proto3,oneof"` } func (*ChasmNexusCompletion_Success) isChasmNexusCompletion_Outcome() {} @@ -849,11 +869,60 @@ func (x *ChasmComponentAttributes_Task) GetPhysicalTaskStatus() int32 { return 0 } +// RequestMetadata groups the framework-tracked metadata contributed by a +// single caller request to this component. +type ChasmComponentAttributes_RequestMetadata struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Links attached by this request. Components use these to record + // references back to the caller(s) that initiated or extended the + // execution (e.g. parent workflow events, Nexus completion sources). + Links []*v1.Link `protobuf:"bytes,1,rep,name=links,proto3" json:"links,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ChasmComponentAttributes_RequestMetadata) Reset() { + *x = ChasmComponentAttributes_RequestMetadata{} + mi := &file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ChasmComponentAttributes_RequestMetadata) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ChasmComponentAttributes_RequestMetadata) ProtoMessage() {} + +func (x *ChasmComponentAttributes_RequestMetadata) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[10] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ChasmComponentAttributes_RequestMetadata.ProtoReflect.Descriptor instead. +func (*ChasmComponentAttributes_RequestMetadata) Descriptor() ([]byte, []int) { + return file_temporal_server_api_persistence_v1_chasm_proto_rawDescGZIP(), []int{2, 1} +} + +func (x *ChasmComponentAttributes_RequestMetadata) GetLinks() []*v1.Link { + if x != nil { + return x.Links + } + return nil +} + var File_temporal_server_api_persistence_v1_chasm_proto protoreflect.FileDescriptor const file_temporal_server_api_persistence_v1_chasm_proto_rawDesc = "" + "\n" + - ".temporal/server/api/persistence/v1/chasm.proto\x12\"temporal.server.api.persistence.v1\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a%temporal/api/failure/v1/message.proto\x1a,temporal/server/api/persistence/v1/hsm.proto\"\x94\x01\n" + + ".temporal/server/api/persistence/v1/chasm.proto\x12\"temporal.server.api.persistence.v1\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a%temporal/api/failure/v1/message.proto\x1a'temporal/api/sdk/v1/user_metadata.proto\x1a,temporal/server/api/persistence/v1/hsm.proto\"\x94\x01\n" + "\tChasmNode\x12Q\n" + "\bmetadata\x18\x01 \x01(\v25.temporal.server.api.persistence.v1.ChasmNodeMetadataR\bmetadata\x124\n" + "\x04data\x18\x02 \x01(\v2 .temporal.api.common.v1.DataBlobR\x04data\"\xd9\x05\n" + @@ -865,13 +934,15 @@ const file_temporal_server_api_persistence_v1_chasm_proto_rawDesc = "" + "\x15collection_attributes\x18\r \x01(\v2=.temporal.server.api.persistence.v1.ChasmCollectionAttributesH\x00R\x14collectionAttributes\x12k\n" + "\x12pointer_attributes\x18\x0e \x01(\v2:.temporal.server.api.persistence.v1.ChasmPointerAttributesH\x00R\x11pointerAttributesB\f\n" + "\n" + - "attributes\"\xbb\x05\n" + + "attributes\"\xbe\b\n" + "\x18ChasmComponentAttributes\x12\x17\n" + "\atype_id\x18\x01 \x01(\rR\x06typeId\x12m\n" + "\x11side_effect_tasks\x18\x02 \x03(\v2A.temporal.server.api.persistence.v1.ChasmComponentAttributes.TaskR\x0fsideEffectTasks\x12`\n" + "\n" + "pure_tasks\x18\x03 \x03(\v2A.temporal.server.api.persistence.v1.ChasmComponentAttributes.TaskR\tpureTasks\x12\x1a\n" + - "\bdetached\x18\x04 \x01(\bR\bdetached\x1a\x98\x03\n" + + "\bdetached\x18\x04 \x01(\bR\bdetached\x12f\n" + + "\brequests\x18\x05 \x03(\v2J.temporal.server.api.persistence.v1.ChasmComponentAttributes.RequestsEntryR\brequests\x12F\n" + + "\ruser_metadata\x18\x06 \x01(\v2!.temporal.api.sdk.v1.UserMetadataR\fuserMetadata\x1a\x98\x03\n" + "\x04Task\x12\x17\n" + "\atype_id\x18\x01 \x01(\rR\x06typeId\x12 \n" + "\vdestination\x18\x02 \x01(\tR\vdestination\x12A\n" + @@ -879,7 +950,12 @@ const file_temporal_server_api_persistence_v1_chasm_proto_rawDesc = "" + "\x04data\x18\x04 \x01(\v2 .temporal.api.common.v1.DataBlobR\x04data\x12j\n" + "\x14versioned_transition\x18\x05 \x01(\v27.temporal.server.api.persistence.v1.VersionedTransitionR\x13versionedTransition\x12>\n" + "\x1bversioned_transition_offset\x18\x06 \x01(\x03R\x19versionedTransitionOffset\x120\n" + - "\x14physical_task_status\x18\a \x01(\x05R\x12physicalTaskStatus\"\x15\n" + + "\x14physical_task_status\x18\a \x01(\x05R\x12physicalTaskStatus\x1aE\n" + + "\x0fRequestMetadata\x122\n" + + "\x05links\x18\x01 \x03(\v2\x1c.temporal.api.common.v1.LinkR\x05links\x1a\x89\x01\n" + + "\rRequestsEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12b\n" + + "\x05value\x18\x02 \x01(\v2L.temporal.server.api.persistence.v1.ChasmComponentAttributes.RequestMetadataR\x05value:\x028\x01\"\x15\n" + "\x13ChasmDataAttributes\"\x1b\n" + "\x19ChasmCollectionAttributes\"5\n" + "\x16ChasmPointerAttributes\x12\x1b\n" + @@ -927,55 +1003,62 @@ func file_temporal_server_api_persistence_v1_chasm_proto_rawDescGZIP() []byte { return file_temporal_server_api_persistence_v1_chasm_proto_rawDescData } -var file_temporal_server_api_persistence_v1_chasm_proto_msgTypes = make([]protoimpl.MessageInfo, 10) +var file_temporal_server_api_persistence_v1_chasm_proto_msgTypes = make([]protoimpl.MessageInfo, 12) var file_temporal_server_api_persistence_v1_chasm_proto_goTypes = []any{ - (*ChasmNode)(nil), // 0: temporal.server.api.persistence.v1.ChasmNode - (*ChasmNodeMetadata)(nil), // 1: temporal.server.api.persistence.v1.ChasmNodeMetadata - (*ChasmComponentAttributes)(nil), // 2: temporal.server.api.persistence.v1.ChasmComponentAttributes - (*ChasmDataAttributes)(nil), // 3: temporal.server.api.persistence.v1.ChasmDataAttributes - (*ChasmCollectionAttributes)(nil), // 4: temporal.server.api.persistence.v1.ChasmCollectionAttributes - (*ChasmPointerAttributes)(nil), // 5: temporal.server.api.persistence.v1.ChasmPointerAttributes - (*ChasmTaskInfo)(nil), // 6: temporal.server.api.persistence.v1.ChasmTaskInfo - (*ChasmComponentRef)(nil), // 7: temporal.server.api.persistence.v1.ChasmComponentRef - (*ChasmNexusCompletion)(nil), // 8: temporal.server.api.persistence.v1.ChasmNexusCompletion - (*ChasmComponentAttributes_Task)(nil), // 9: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task - (*v1.DataBlob)(nil), // 10: temporal.api.common.v1.DataBlob - (*VersionedTransition)(nil), // 11: temporal.server.api.persistence.v1.VersionedTransition - (*v1.Payload)(nil), // 12: temporal.api.common.v1.Payload - (*v11.Failure)(nil), // 13: temporal.api.failure.v1.Failure - (*timestamppb.Timestamp)(nil), // 14: google.protobuf.Timestamp - (*v1.Link)(nil), // 15: temporal.api.common.v1.Link + (*ChasmNode)(nil), // 0: temporal.server.api.persistence.v1.ChasmNode + (*ChasmNodeMetadata)(nil), // 1: temporal.server.api.persistence.v1.ChasmNodeMetadata + (*ChasmComponentAttributes)(nil), // 2: temporal.server.api.persistence.v1.ChasmComponentAttributes + (*ChasmDataAttributes)(nil), // 3: temporal.server.api.persistence.v1.ChasmDataAttributes + (*ChasmCollectionAttributes)(nil), // 4: temporal.server.api.persistence.v1.ChasmCollectionAttributes + (*ChasmPointerAttributes)(nil), // 5: temporal.server.api.persistence.v1.ChasmPointerAttributes + (*ChasmTaskInfo)(nil), // 6: temporal.server.api.persistence.v1.ChasmTaskInfo + (*ChasmComponentRef)(nil), // 7: temporal.server.api.persistence.v1.ChasmComponentRef + (*ChasmNexusCompletion)(nil), // 8: temporal.server.api.persistence.v1.ChasmNexusCompletion + (*ChasmComponentAttributes_Task)(nil), // 9: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task + (*ChasmComponentAttributes_RequestMetadata)(nil), // 10: temporal.server.api.persistence.v1.ChasmComponentAttributes.RequestMetadata + nil, // 11: temporal.server.api.persistence.v1.ChasmComponentAttributes.RequestsEntry + (*v1.DataBlob)(nil), // 12: temporal.api.common.v1.DataBlob + (*VersionedTransition)(nil), // 13: temporal.server.api.persistence.v1.VersionedTransition + (*v11.UserMetadata)(nil), // 14: temporal.api.sdk.v1.UserMetadata + (*v1.Payload)(nil), // 15: temporal.api.common.v1.Payload + (*v12.Failure)(nil), // 16: temporal.api.failure.v1.Failure + (*timestamppb.Timestamp)(nil), // 17: google.protobuf.Timestamp + (*v1.Link)(nil), // 18: temporal.api.common.v1.Link } var file_temporal_server_api_persistence_v1_chasm_proto_depIdxs = []int32{ 1, // 0: temporal.server.api.persistence.v1.ChasmNode.metadata:type_name -> temporal.server.api.persistence.v1.ChasmNodeMetadata - 10, // 1: temporal.server.api.persistence.v1.ChasmNode.data:type_name -> temporal.api.common.v1.DataBlob - 11, // 2: temporal.server.api.persistence.v1.ChasmNodeMetadata.initial_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 11, // 3: temporal.server.api.persistence.v1.ChasmNodeMetadata.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 12, // 1: temporal.server.api.persistence.v1.ChasmNode.data:type_name -> temporal.api.common.v1.DataBlob + 13, // 2: temporal.server.api.persistence.v1.ChasmNodeMetadata.initial_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 13, // 3: temporal.server.api.persistence.v1.ChasmNodeMetadata.last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition 2, // 4: temporal.server.api.persistence.v1.ChasmNodeMetadata.component_attributes:type_name -> temporal.server.api.persistence.v1.ChasmComponentAttributes 3, // 5: temporal.server.api.persistence.v1.ChasmNodeMetadata.data_attributes:type_name -> temporal.server.api.persistence.v1.ChasmDataAttributes 4, // 6: temporal.server.api.persistence.v1.ChasmNodeMetadata.collection_attributes:type_name -> temporal.server.api.persistence.v1.ChasmCollectionAttributes 5, // 7: temporal.server.api.persistence.v1.ChasmNodeMetadata.pointer_attributes:type_name -> temporal.server.api.persistence.v1.ChasmPointerAttributes 9, // 8: temporal.server.api.persistence.v1.ChasmComponentAttributes.side_effect_tasks:type_name -> temporal.server.api.persistence.v1.ChasmComponentAttributes.Task 9, // 9: temporal.server.api.persistence.v1.ChasmComponentAttributes.pure_tasks:type_name -> temporal.server.api.persistence.v1.ChasmComponentAttributes.Task - 11, // 10: temporal.server.api.persistence.v1.ChasmTaskInfo.component_initial_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 11, // 11: temporal.server.api.persistence.v1.ChasmTaskInfo.component_last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 10, // 12: temporal.server.api.persistence.v1.ChasmTaskInfo.data:type_name -> temporal.api.common.v1.DataBlob - 11, // 13: temporal.server.api.persistence.v1.ChasmTaskInfo.task_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 11, // 14: temporal.server.api.persistence.v1.ChasmComponentRef.execution_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 11, // 15: temporal.server.api.persistence.v1.ChasmComponentRef.component_initial_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 12, // 16: temporal.server.api.persistence.v1.ChasmNexusCompletion.success:type_name -> temporal.api.common.v1.Payload - 13, // 17: temporal.server.api.persistence.v1.ChasmNexusCompletion.failure:type_name -> temporal.api.failure.v1.Failure - 14, // 18: temporal.server.api.persistence.v1.ChasmNexusCompletion.close_time:type_name -> google.protobuf.Timestamp - 15, // 19: temporal.server.api.persistence.v1.ChasmNexusCompletion.links:type_name -> temporal.api.common.v1.Link - 14, // 20: temporal.server.api.persistence.v1.ChasmNexusCompletion.start_time:type_name -> google.protobuf.Timestamp - 14, // 21: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.scheduled_time:type_name -> google.protobuf.Timestamp - 10, // 22: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.data:type_name -> temporal.api.common.v1.DataBlob - 11, // 23: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition - 24, // [24:24] is the sub-list for method output_type - 24, // [24:24] is the sub-list for method input_type - 24, // [24:24] is the sub-list for extension type_name - 24, // [24:24] is the sub-list for extension extendee - 0, // [0:24] is the sub-list for field type_name + 11, // 10: temporal.server.api.persistence.v1.ChasmComponentAttributes.requests:type_name -> temporal.server.api.persistence.v1.ChasmComponentAttributes.RequestsEntry + 14, // 11: temporal.server.api.persistence.v1.ChasmComponentAttributes.user_metadata:type_name -> temporal.api.sdk.v1.UserMetadata + 13, // 12: temporal.server.api.persistence.v1.ChasmTaskInfo.component_initial_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 13, // 13: temporal.server.api.persistence.v1.ChasmTaskInfo.component_last_update_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 12, // 14: temporal.server.api.persistence.v1.ChasmTaskInfo.data:type_name -> temporal.api.common.v1.DataBlob + 13, // 15: temporal.server.api.persistence.v1.ChasmTaskInfo.task_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 13, // 16: temporal.server.api.persistence.v1.ChasmComponentRef.execution_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 13, // 17: temporal.server.api.persistence.v1.ChasmComponentRef.component_initial_versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 15, // 18: temporal.server.api.persistence.v1.ChasmNexusCompletion.success:type_name -> temporal.api.common.v1.Payload + 16, // 19: temporal.server.api.persistence.v1.ChasmNexusCompletion.failure:type_name -> temporal.api.failure.v1.Failure + 17, // 20: temporal.server.api.persistence.v1.ChasmNexusCompletion.close_time:type_name -> google.protobuf.Timestamp + 18, // 21: temporal.server.api.persistence.v1.ChasmNexusCompletion.links:type_name -> temporal.api.common.v1.Link + 17, // 22: temporal.server.api.persistence.v1.ChasmNexusCompletion.start_time:type_name -> google.protobuf.Timestamp + 17, // 23: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.scheduled_time:type_name -> google.protobuf.Timestamp + 12, // 24: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.data:type_name -> temporal.api.common.v1.DataBlob + 13, // 25: temporal.server.api.persistence.v1.ChasmComponentAttributes.Task.versioned_transition:type_name -> temporal.server.api.persistence.v1.VersionedTransition + 18, // 26: temporal.server.api.persistence.v1.ChasmComponentAttributes.RequestMetadata.links:type_name -> temporal.api.common.v1.Link + 10, // 27: temporal.server.api.persistence.v1.ChasmComponentAttributes.RequestsEntry.value:type_name -> temporal.server.api.persistence.v1.ChasmComponentAttributes.RequestMetadata + 28, // [28:28] is the sub-list for method output_type + 28, // [28:28] is the sub-list for method input_type + 28, // [28:28] is the sub-list for extension type_name + 28, // [28:28] is the sub-list for extension extendee + 0, // [0:28] is the sub-list for field type_name } func init() { file_temporal_server_api_persistence_v1_chasm_proto_init() } @@ -1000,7 +1083,7 @@ func file_temporal_server_api_persistence_v1_chasm_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc), len(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc)), NumEnums: 0, - NumMessages: 10, + NumMessages: 12, NumExtensions: 0, NumServices: 0, }, diff --git a/chasm/context.go b/chasm/context.go index ebf4bee1a4..87e057f541 100644 --- a/chasm/context.go +++ b/chasm/context.go @@ -5,6 +5,8 @@ import ( "errors" "time" + commonpb "go.temporal.io/api/common/v1" + sdkpb "go.temporal.io/api/sdk/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" @@ -46,6 +48,14 @@ type Context interface { // callbacks invoked by the chasm engine. In other contexts, such as pure tasks executed at the end of a transaction // or background task handlers, the underlying ctx has no gRPC metadata and this method always returns "". RequestHeader(key string) string + // Links returns the union of links attached to the given component across all requests. + // Returns nil for components that are not (yet) registered as tree nodes. + Links(Component) []*commonpb.Link + // RequestLinks returns the links attached to the given component for the specific requestID. + // Returns nil if no entry exists for that requestID. Empty requestID is rejected. + RequestLinks(Component, string) ([]*commonpb.Link, error) + // UserMetadata returns the user metadata attached to the given component, or nil if none. + UserMetadata(Component) *sdkpb.UserMetadata // Intent() OperationIntent // ComponentOptions(Component) []ComponentOption @@ -80,6 +90,12 @@ type MutableContext interface { // The task is associated with the given component and will be invoked via the registered handler for the given task // referencing the component. AddTask(Component, TaskAttributes, any) + // SetRequestLinks records the links contributed by the given request on the + // component, replacing any prior entry for the same request ID. Passing + // nil/empty links removes the entry. + SetRequestLinks(Component, string, []*commonpb.Link) error + // SetUserMetadata replaces the user metadata attached to the given component. + SetUserMetadata(Component, *sdkpb.UserMetadata) error // Get a Ref for the component // This ref to the component state at the end of the transition @@ -140,6 +156,18 @@ func (c *immutableCtx) Ref(component Component) ([]byte, error) { return c.root.Ref(component) } +func (c *immutableCtx) Links(component Component) []*commonpb.Link { + return c.root.componentLinks(component) +} + +func (c *immutableCtx) RequestLinks(component Component, requestID string) ([]*commonpb.Link, error) { + return c.root.componentRequestLinks(component, requestID) +} + +func (c *immutableCtx) UserMetadata(component Component) *sdkpb.UserMetadata { + return c.root.componentUserMetadata(component) +} + func (c *immutableCtx) Now(component Component) time.Time { return c.root.Now(component) } @@ -236,6 +264,14 @@ func (c *mutableCtx) AddTask( c.root.AddTask(component, attributes, payload) } +func (c *mutableCtx) SetRequestLinks(component Component, requestID string, links []*commonpb.Link) error { + return c.root.setComponentRequestLinks(component, requestID, links) +} + +func (c *mutableCtx) SetUserMetadata(component Component, md *sdkpb.UserMetadata) error { + return c.root.setComponentUserMetadata(component, md) +} + func (c *mutableCtx) withValue(key any, value any) Context { return &mutableCtx{ immutableCtx: ContextWithValue(c.immutableCtx, key, value), diff --git a/chasm/context_mock.go b/chasm/context_mock.go index b9f79e1fbe..92b05f780b 100644 --- a/chasm/context_mock.go +++ b/chasm/context_mock.go @@ -8,6 +8,8 @@ import ( "sync" "time" + commonpb "go.temporal.io/api/common/v1" + sdkpb "go.temporal.io/api/sdk/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/common/log" "go.temporal.io/server/common/log/tag" @@ -31,6 +33,9 @@ type MockContext struct { HandleNamespaceEntry func() *namespace.Namespace HandleEndpointByName func(string) (*persistencespb.NexusEndpointEntry, error) HandleMetricsHandler func() metrics.Handler + HandleLinks func(component Component) []*commonpb.Link + HandleRequestLinks func(component Component, requestID string) ([]*commonpb.Link, error) + HandleUserMetadata func(component Component) *sdkpb.UserMetadata // GoCtx is the underlying context.Context used for context value lookups. // Any values set on it will be available via the CHASM mock context's Value method, @@ -136,6 +141,27 @@ func (c *MockContext) Value(key any) any { return c.goContext().Value(key) } +func (c *MockContext) Links(component Component) []*commonpb.Link { + if c.HandleLinks != nil { + return c.HandleLinks(component) + } + return nil +} + +func (c *MockContext) RequestLinks(component Component, requestID string) ([]*commonpb.Link, error) { + if c.HandleRequestLinks != nil { + return c.HandleRequestLinks(component, requestID) + } + return nil, nil +} + +func (c *MockContext) UserMetadata(component Component) *sdkpb.UserMetadata { + if c.HandleUserMetadata != nil { + return c.HandleUserMetadata(component) + } + return nil +} + func (c *MockContext) withValue(key any, value any) Context { return &MockContext{ HandleExecutionKey: c.HandleExecutionKey, @@ -146,6 +172,9 @@ func (c *MockContext) withValue(key any, value any) Context { GoCtx: context.WithValue(c.goContext(), key, value), HandleNamespaceEntry: c.HandleNamespaceEntry, HandleEndpointByName: c.HandleEndpointByName, + HandleLinks: c.HandleLinks, + HandleRequestLinks: c.HandleRequestLinks, + HandleUserMetadata: c.HandleUserMetadata, } } @@ -154,8 +183,10 @@ func (c *MockContext) withValue(key any, value any) Context { type MockMutableContext struct { MockContext - mu sync.Mutex - Tasks []MockTask + mu sync.Mutex + Tasks []MockTask + LinksByRequest map[Component]map[string][]*commonpb.Link + UserMetadataByComponent map[Component]*sdkpb.UserMetadata } func (c *MockMutableContext) AddTask(component Component, attributes TaskAttributes, payload any) { @@ -164,6 +195,35 @@ func (c *MockMutableContext) AddTask(component Component, attributes TaskAttribu c.Tasks = append(c.Tasks, MockTask{component, attributes, payload}) } +func (c *MockMutableContext) SetRequestLinks(component Component, requestID string, links []*commonpb.Link) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.LinksByRequest == nil { + c.LinksByRequest = make(map[Component]map[string][]*commonpb.Link) + } + perRequest, ok := c.LinksByRequest[component] + if !ok { + perRequest = make(map[string][]*commonpb.Link) + c.LinksByRequest[component] = perRequest + } + if len(links) == 0 { + delete(perRequest, requestID) + } else { + perRequest[requestID] = links + } + return nil +} + +func (c *MockMutableContext) SetUserMetadata(component Component, md *sdkpb.UserMetadata) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.UserMetadataByComponent == nil { + c.UserMetadataByComponent = make(map[Component]*sdkpb.UserMetadata) + } + c.UserMetadataByComponent[component] = md + return nil +} + func (c *MockMutableContext) withValue(key any, value any) Context { return &MockMutableContext{ MockContext: *ContextWithValue(&c.MockContext, key, value), diff --git a/chasm/tree.go b/chasm/tree.go index 8b85b3a783..03ce086c29 100644 --- a/chasm/tree.go +++ b/chasm/tree.go @@ -15,6 +15,7 @@ import ( commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" historypb "go.temporal.io/api/history/v1" + sdkpb "go.temporal.io/api/sdk/v1" "go.temporal.io/api/serviceerror" enumsspb "go.temporal.io/server/api/enums/v1" persistencespb "go.temporal.io/server/api/persistence/v1" @@ -148,6 +149,12 @@ type ( newTasks map[any][]taskWithAttributes // component value -> task & attributes immediatePureTasks map[any][]taskWithAttributes // similar to newTasks, but will be executed at the end of the transaction + // Pending framework metadata writes keyed by component value. Applied to + // each component's ChasmComponentAttributes during CloseTransaction so + // callers can stage writes before the component is registered as a node. + pendingRequestLinks map[any]map[string][]*commonpb.Link + pendingUserMetadata map[any]*sdkpb.UserMetadata + // Node value -> node // Only component and data node values are tracked right now valueToNode map[any]*Node @@ -330,6 +337,8 @@ func newTreeHelper( }, newTasks: make(map[any][]taskWithAttributes), immediatePureTasks: make(map[any][]taskWithAttributes), + pendingRequestLinks: make(map[any]map[string][]*commonpb.Link), + pendingUserMetadata: make(map[any]*sdkpb.UserMetadata), valueToNode: make(map[any]*Node), taskValueCache: make(map[*commonpb.DataBlob]reflect.Value), needsPointerResolution: false, @@ -1431,6 +1440,164 @@ func (n *Node) structuredRef( } +// componentLinks returns the union of links across all requests stored on the +// given component's metadata. Pending writes staged in the current transaction +// replace persisted entries for the same request ID (matching the read +// semantics of componentRequestLinks), so a caller staging an update doesn't +// observe stale + new entries side-by-side. +func (n *Node) componentLinks(component Component) []*commonpb.Link { + var links []*commonpb.Link + pending := n.pendingRequestLinks[component] + + for _, ls := range pending { + links = append(links, ls...) + } + + if refNode, ok := n.valueToNode[component]; ok && refNode.isComponent() { + for requestID, req := range refNode.serializedNode.GetMetadata().GetComponentAttributes().GetRequests() { + if _, overridden := pending[requestID]; overridden { + continue + } + links = append(links, req.GetLinks()...) + } + } + + return links +} + +// setComponentRequestLinks records the links contributed by the given request +// on the component, replacing any prior entry for the same request ID. Passing +// nil/empty links removes the entry. The write is staged and applied during +// CloseTransaction, so it works for components that have not yet been +// registered as nodes. An empty requestID is rejected to avoid silent +// collisions across callers. +func (n *Node) setComponentRequestLinks(component Component, requestID string, links []*commonpb.Link) error { + if requestID == "" { + return serviceerror.NewInvalidArgument("requestID is required when setting per-request links") + } + perRequest, ok := n.pendingRequestLinks[component] + if !ok { + perRequest = make(map[string][]*commonpb.Link) + n.pendingRequestLinks[component] = perRequest + } + perRequest[requestID] = links + return nil +} + +// componentRequestLinks returns the links stored on the given component's +// metadata for the specific requestID, preferring a pending write staged in +// the current transaction. Returns nil if no entry exists. +func (n *Node) componentRequestLinks(component Component, requestID string) ([]*commonpb.Link, error) { + if requestID == "" { + return nil, serviceerror.NewInvalidArgument("requestID is required when reading per-request links") + } + if pending, ok := n.pendingRequestLinks[component]; ok { + if links, ok := pending[requestID]; ok { + return links, nil + } + } + if refNode, ok := n.valueToNode[component]; ok && refNode.isComponent() { + if req, ok := refNode.serializedNode.GetMetadata().GetComponentAttributes().GetRequests()[requestID]; ok { + return req.GetLinks(), nil + } + } + return nil, nil +} + +// componentUserMetadata returns the user metadata stored on the given +// component, preferring a pending write staged in the current transaction. +func (n *Node) componentUserMetadata(component Component) *sdkpb.UserMetadata { + if md, ok := n.pendingUserMetadata[component]; ok { + return md + } + if refNode, ok := n.valueToNode[component]; ok && refNode.isComponent() { + return refNode.serializedNode.GetMetadata().GetComponentAttributes().GetUserMetadata() + } + return nil +} + +// setComponentUserMetadata stages a user-metadata write for the given component. +// Applied during CloseTransaction. +func (n *Node) setComponentUserMetadata(component Component, md *sdkpb.UserMetadata) error { + n.pendingUserMetadata[component] = md + return nil +} + +// closeTransactionApplyPendingComponentMetadata walks the tree and applies any +// staged framework metadata (request links, user metadata) to each component +// node, marking touched nodes as updated for replication. Pending entries that +// reference a component that was never registered as a node are dropped and +// logged at warn level to surface caller misuse. +func (n *Node) closeTransactionApplyPendingComponentMetadata() error { + if len(n.pendingRequestLinks) == 0 && len(n.pendingUserMetadata) == 0 { + return nil + } + for _, node := range n.andAllChildren() { + if !node.applyPendingComponentMetadata() { + continue + } + encodedPath, err := node.getEncodedPath() + if err != nil { + return err + } + if _, exists := n.mutation.UpdatedNodes[encodedPath]; !exists { + node.updateLastUpdateVersionedTransition() + n.mutation.UpdatedNodes[encodedPath] = node.serializedNode + delete(n.mutation.DeletedNodes, encodedPath) + } + } + if len(n.pendingRequestLinks) > 0 || len(n.pendingUserMetadata) > 0 { + n.logger.Warn( + "chasm: dropped staged component metadata for components that were never registered as nodes", + tag.NewInt("orphan-request-link-components", len(n.pendingRequestLinks)), + tag.NewInt("orphan-user-metadata-components", len(n.pendingUserMetadata)), + ) + } + n.pendingRequestLinks = make(map[any]map[string][]*commonpb.Link) + n.pendingUserMetadata = make(map[any]*sdkpb.UserMetadata) + return nil +} + +// applyPendingComponentMetadata writes staged per-component framework metadata +// (request links and user metadata) onto the node's ChasmComponentAttributes. +// Returns true if the node was mutated. +func (n *Node) applyPendingComponentMetadata() bool { + if n.value == nil { + return false + } + attrs := n.serializedNode.GetMetadata().GetComponentAttributes() + if attrs == nil { + return false + } + dirty := false + + if pending, ok := n.pendingRequestLinks[n.value]; ok { + if attrs.Requests == nil && len(pending) > 0 { + attrs.Requests = make(map[string]*persistencespb.ChasmComponentAttributes_RequestMetadata) + } + for requestID, links := range pending { + if len(links) == 0 { + if _, exists := attrs.Requests[requestID]; exists { + delete(attrs.Requests, requestID) + dirty = true + } + continue + } + attrs.Requests[requestID] = &persistencespb.ChasmComponentAttributes_RequestMetadata{Links: links} + dirty = true + } + delete(n.pendingRequestLinks, n.value) + } + + if md, ok := n.pendingUserMetadata[n.value]; ok { + attrs.UserMetadata = md + dirty = true + delete(n.pendingUserMetadata, n.value) + } + + return dirty +} + // componentNodePath implements the CHASM Context interface func (n *Node) componentNodePath( component Component, @@ -1535,6 +1702,10 @@ func (n *Node) CloseTransaction() (NodesMutation, error) { return NodesMutation{}, err } + if err := n.closeTransactionApplyPendingComponentMetadata(); err != nil { + return NodesMutation{}, err + } + // Both user & system data mutation need to be returned and persisted. maps.Copy(n.mutation.UpdatedNodes, n.systemMutation.UpdatedNodes) maps.Copy(n.mutation.DeletedNodes, n.systemMutation.DeletedNodes) @@ -2248,6 +2419,13 @@ func (n *Node) cleanupTransaction() { n.immediatePureTasks = make(map[any][]taskWithAttributes) } + if len(n.pendingRequestLinks) != 0 { + n.pendingRequestLinks = make(map[any]map[string][]*commonpb.Link) + } + if len(n.pendingUserMetadata) != 0 { + n.pendingUserMetadata = make(map[any]*sdkpb.UserMetadata) + } + n.isActiveStateDirty = false n.needsPointerResolution = false } @@ -2685,7 +2863,9 @@ func (n *Node) IsDirty() bool { // which need to be persisted to DB AND replicated to other clusters. // The result will be reset to false after a call to CloseTransaction(). func (n *Node) IsStateDirty() bool { - return n.isActiveStateDirty || len(n.mutation.UpdatedNodes) > 0 || len(n.mutation.DeletedNodes) > 0 + return n.isActiveStateDirty || + len(n.mutation.UpdatedNodes) > 0 || + len(n.mutation.DeletedNodes) > 0 } func (n *Node) IsStale( diff --git a/chasm/tree_test.go b/chasm/tree_test.go index 506e2fb4fd..af6e1865b4 100644 --- a/chasm/tree_test.go +++ b/chasm/tree_test.go @@ -15,6 +15,7 @@ import ( "github.com/stretchr/testify/suite" commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" + sdkpb "go.temporal.io/api/sdk/v1" "go.temporal.io/api/serviceerror" enumsspb "go.temporal.io/server/api/enums/v1" persistencespb "go.temporal.io/server/api/persistence/v1" @@ -4049,3 +4050,361 @@ func (e protoMatcher) String() string { func protoEq(x proto.Message) gomock.Matcher { return protoMatcher{x: x} } + +// TestCloseTransaction_AppliesPendingComponentMetadata verifies that +// SetRequestLinks/SetUserMetadata writes are written onto the root component's +// ChasmComponentAttributes during CloseTransaction, that the touched node is +// added to NodesMutation.UpdatedNodes, and that its LastUpdateVersionedTransition +// is bumped. +func (s *nodeSuite) TestCloseTransaction_AppliesPendingComponentMetadata() { + const requestID = "req-1" + link := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"}, + }} + md := &sdkpb.UserMetadata{Summary: &commonpb.Payload{Data: []byte("summary")}} + + root := s.testComponentTree() // sets HandleNextTransitionCount = 1, HandleGetCurrentVersion = 1 + + // Initial create transaction must close cleanly before we exercise the metadata path. + _, err := root.CloseTransaction() + s.NoError(err) + + // Bump the transition count so we can verify LastUpdateVersionedTransition was updated. + s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 } + + ctx := NewMutableContext(context.Background(), root) + c, err := root.Component(ctx, ComponentRef{}) + s.NoError(err) + + s.NoError(ctx.SetRequestLinks(c, requestID, []*commonpb.Link{link})) + s.NoError(ctx.SetUserMetadata(c, md)) + + mutation, err := root.CloseTransaction() + s.NoError(err) + + rootSerialized, ok := mutation.UpdatedNodes[""] + s.True(ok, "root node must appear in UpdatedNodes after staging metadata") + attrs := rootSerialized.GetMetadata().GetComponentAttributes() + s.NotNil(attrs) + s.Equal([]*commonpb.Link{link}, attrs.GetRequests()[requestID].GetLinks()) + s.Equal(md.GetSummary().GetData(), attrs.GetUserMetadata().GetSummary().GetData()) + s.Equal(int64(2), rootSerialized.GetMetadata().GetLastUpdateVersionedTransition().GetTransitionCount()) + + // Pending maps must be cleared after CloseTransaction so a subsequent transaction + // does not re-apply the same writes. + s.Empty(root.pendingRequestLinks) + s.Empty(root.pendingUserMetadata) +} + +// TestSetComponentMetadata_MarksTreeDirty verifies that staging a +// SetRequestLinks or SetUserMetadata write flips IsDirty()/IsStateDirty() +// before CloseTransaction runs. +func (s *nodeSuite) TestSetComponentMetadata_MarksTreeDirty() { + root := s.testComponentTree() + _, err := root.CloseTransaction() + s.NoError(err) + + s.False(root.IsDirty(), "tree must be clean after the initial close") + s.False(root.IsStateDirty()) + + ctx := NewMutableContext(context.Background(), root) + c, err := root.Component(ctx, ComponentRef{}) + s.NoError(err) + + s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{{ + Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"}, + }, + }})) + s.True(root.IsStateDirty(), "staging SetRequestLinks must mark the tree dirty") + s.True(root.IsDirty()) + + _, err = root.CloseTransaction() + s.NoError(err) + s.False(root.IsStateDirty(), "CloseTransaction must clear the dirty flag") + + ctx = NewMutableContext(context.Background(), root) + c, err = root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetUserMetadata(c, &sdkpb.UserMetadata{ + Summary: &commonpb.Payload{Data: []byte("summary")}, + })) + s.True(root.IsStateDirty(), "staging SetUserMetadata must mark the tree dirty") + s.True(root.IsDirty()) +} + +// TestCloseTransaction_DropsOrphanedComponentMetadata verifies that pending +// SetRequestLinks/SetUserMetadata writes against a component value that is +// not registered in the tree are silently dropped during CloseTransaction +// (rather than panicking or surfacing an error), and that the pending maps +// are cleared afterwards. +func (s *nodeSuite) TestCloseTransaction_DropsOrphanedComponentMetadata() { + root := s.testComponentTree() + _, err := root.CloseTransaction() + s.NoError(err) + + s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 } + ctx := NewMutableContext(context.Background(), root) + + // Stage writes against a component value that was never set on the tree. + orphan := &TestComponent{} + s.NoError(ctx.SetRequestLinks(orphan, "req-id", []*commonpb.Link{{ + Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"}, + }, + }})) + s.NoError(ctx.SetUserMetadata(orphan, &sdkpb.UserMetadata{ + Summary: &commonpb.Payload{Data: []byte("orphan")}, + })) + + mutation, err := root.CloseTransaction() + s.NoError(err) + s.NotContains(mutation.UpdatedNodes, "", "root must not be updated by orphaned writes") + s.Empty(root.pendingRequestLinks) + s.Empty(root.pendingUserMetadata) +} + +// TestSetComponentRequestLinks_RejectsEmptyRequestID verifies the framework +// hard-rejects empty requestIDs so two callers cannot silently collide on the +// empty-string key. +func (s *nodeSuite) TestSetComponentRequestLinks_RejectsEmptyRequestID() { + root := s.testComponentTree() + _, err := root.CloseTransaction() + s.NoError(err) + + ctx := NewMutableContext(context.Background(), root) + c, err := root.Component(ctx, ComponentRef{}) + s.NoError(err) + + err = ctx.SetRequestLinks(c, "", []*commonpb.Link{{ + Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"}, + }, + }}) + s.Error(err) + s.ErrorAs(err, new(*serviceerror.InvalidArgument)) + + _, err = ctx.RequestLinks(c, "") + s.Error(err) + s.ErrorAs(err, new(*serviceerror.InvalidArgument)) +} + +// TestSetRequestLinks_MultipleRequestsCoexist verifies that two distinct +// request IDs on the same component land as separate entries in +// ChasmComponentAttributes.requests. +func (s *nodeSuite) TestSetRequestLinks_MultipleRequestsCoexist() { + root := s.testComponentTree() + _, err := root.CloseTransaction() + s.NoError(err) + + s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 } + ctx := NewMutableContext(context.Background(), root) + c, err := root.Component(ctx, ComponentRef{}) + s.NoError(err) + + linkA := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "a", RunId: "run"}, + }} + linkB := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "b", RunId: "run"}, + }} + s.NoError(ctx.SetRequestLinks(c, "req-a", []*commonpb.Link{linkA})) + s.NoError(ctx.SetRequestLinks(c, "req-b", []*commonpb.Link{linkB})) + + mutation, err := root.CloseTransaction() + s.NoError(err) + + attrs := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes() + s.Len(attrs.GetRequests(), 2) + s.Equal([]*commonpb.Link{linkA}, attrs.GetRequests()["req-a"].GetLinks()) + s.Equal([]*commonpb.Link{linkB}, attrs.GetRequests()["req-b"].GetLinks()) +} + +// TestSetRequestLinks_ReplacesEntryForSameRequestID verifies that two +// SetRequestLinks calls with the same requestID — within or across +// transactions — leave only the second value in attrs.Requests. +func (s *nodeSuite) TestSetRequestLinks_ReplacesEntryForSameRequestID() { + root := s.testComponentTree() + _, err := root.CloseTransaction() + s.NoError(err) + + linkA := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "a", RunId: "run"}, + }} + linkB := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "b", RunId: "run"}, + }} + + // Within a single transaction: second SetRequestLinks for the same requestID + // must overwrite the first. + nextTC := int64(2) + s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC } + ctx := NewMutableContext(context.Background(), root) + c, err := root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{linkA})) + s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{linkB})) + mutation, err := root.CloseTransaction() + s.NoError(err) + s.Equal([]*commonpb.Link{linkB}, + mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes().GetRequests()["req"].GetLinks(), + "second SetRequestLinks within a transaction must overwrite the first", + ) + + // Across transactions: a later write for the same requestID must replace the + // previously persisted entry. + nextTC = 3 + ctx = NewMutableContext(context.Background(), root) + c, err = root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{linkA})) + mutation, err = root.CloseTransaction() + s.NoError(err) + s.Equal([]*commonpb.Link{linkA}, + mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes().GetRequests()["req"].GetLinks(), + "a follow-up transaction's SetRequestLinks must replace the persisted entry", + ) +} + +// TestSetRequestLinks_RemovesEntryWhenEmptyLinks verifies that passing nil/empty +// links for a previously-stored requestID removes that entry from +// ChasmComponentAttributes.requests. +func (s *nodeSuite) TestSetRequestLinks_RemovesEntryWhenEmptyLinks() { + root := s.testComponentTree() + _, err := root.CloseTransaction() + s.NoError(err) + + link := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"}, + }} + + // First persist a link under "req". + nextTC := int64(2) + s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC } + ctx := NewMutableContext(context.Background(), root) + c, err := root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{link})) + mutation, err := root.CloseTransaction() + s.NoError(err) + s.Contains(mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes().GetRequests(), "req") + + // Then clear it with an empty links slice. + nextTC = 3 + ctx = NewMutableContext(context.Background(), root) + c, err = root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetRequestLinks(c, "req", nil)) + mutation, err = root.CloseTransaction() + s.NoError(err) + attrs := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes() + s.NotContains(attrs.GetRequests(), "req", "empty links must remove the entry for requestID") +} + +// TestRequestLinks_PrefersPendingOverPersisted verifies that an in-transaction +// SetRequestLinks shadow-reads via RequestLinks / Links return the staged +// (pending) value rather than the previously-persisted entry, so callers +// reading-then-writing within a single transaction never observe stale state. +func (s *nodeSuite) TestRequestLinks_PrefersPendingOverPersisted() { + root := s.testComponentTree() + _, err := root.CloseTransaction() + s.NoError(err) + + oldLink := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "old", RunId: "run"}, + }} + newLink := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "new", RunId: "run"}, + }} + + // Persist [oldLink] under "req". + nextTC := int64(2) + s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC } + ctx := NewMutableContext(context.Background(), root) + c, err := root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{oldLink})) + _, err = root.CloseTransaction() + s.NoError(err) + + // Open a new transaction, stage a replace with [newLink] under the same + // requestID, then read via both APIs before close. + nextTC = 3 + ctx = NewMutableContext(context.Background(), root) + c, err = root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{newLink})) + + got, err := ctx.RequestLinks(c, "req") + s.NoError(err) + s.Equal([]*commonpb.Link{newLink}, got, "RequestLinks must prefer pending over persisted for the same requestID") + s.Equal([]*commonpb.Link{newLink}, ctx.Links(c), + "Links must prefer pending and not return old+new duplicates for the same requestID") +} + +// TestCloseTransaction_PersistsAcrossTransactions verifies the realistic +// production flow: write metadata in transaction A, commit, open transaction +// B, read it back through the framework APIs. +func (s *nodeSuite) TestCloseTransaction_PersistsAcrossTransactions() { + root := s.testComponentTree() + _, err := root.CloseTransaction() + s.NoError(err) + + link := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{ + WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"}, + }} + md := &sdkpb.UserMetadata{Summary: &commonpb.Payload{Data: []byte("summary")}} + + nextTC := int64(2) + s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC } + ctx := NewMutableContext(context.Background(), root) + c, err := root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{link})) + s.NoError(ctx.SetUserMetadata(c, md)) + _, err = root.CloseTransaction() + s.NoError(err) + + // New transaction: framework getters must surface the persisted attrs. + nextTC = 3 + ctx2 := NewMutableContext(context.Background(), root) + c2, err := root.Component(ctx2, ComponentRef{}) + s.NoError(err) + + got, err := ctx2.RequestLinks(c2, "req") + s.NoError(err) + s.Equal([]*commonpb.Link{link}, got) + s.Equal([]*commonpb.Link{link}, ctx2.Links(c2)) + s.ProtoEqual(md, ctx2.UserMetadata(c2)) +} + +// TestSetUserMetadata_NilClearsPersistedValue verifies that SetUserMetadata +// called with nil clears any previously-persisted user metadata on the +// component (rather than being treated as a no-op). +func (s *nodeSuite) TestSetUserMetadata_NilClearsPersistedValue() { + root := s.testComponentTree() + _, err := root.CloseTransaction() + s.NoError(err) + + // Persist user metadata. + nextTC := int64(2) + s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC } + ctx := NewMutableContext(context.Background(), root) + c, err := root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetUserMetadata(c, &sdkpb.UserMetadata{ + Summary: &commonpb.Payload{Data: []byte("first")}, + })) + _, err = root.CloseTransaction() + s.NoError(err) + + // Clear with nil. + nextTC = 3 + ctx = NewMutableContext(context.Background(), root) + c, err = root.Component(ctx, ComponentRef{}) + s.NoError(err) + s.NoError(ctx.SetUserMetadata(c, nil)) + mutation, err := root.CloseTransaction() + s.NoError(err) + s.Nil(mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes().GetUserMetadata()) +} diff --git a/proto/internal/temporal/server/api/persistence/v1/chasm.proto b/proto/internal/temporal/server/api/persistence/v1/chasm.proto index cff53f278b..290db15d07 100644 --- a/proto/internal/temporal/server/api/persistence/v1/chasm.proto +++ b/proto/internal/temporal/server/api/persistence/v1/chasm.proto @@ -5,6 +5,7 @@ package temporal.server.api.persistence.v1; import "google/protobuf/timestamp.proto"; import "temporal/api/common/v1/message.proto"; import "temporal/api/failure/v1/message.proto"; +import "temporal/api/sdk/v1/user_metadata.proto"; import "temporal/server/api/persistence/v1/hsm.proto"; option go_package = "go.temporal.io/server/api/persistence/v1;persistence"; @@ -50,6 +51,14 @@ message ChasmComponentAttributes { // Changes to this field also doesn't require an increase in versioned transition. int32 physical_task_status = 7; } + // RequestMetadata groups the framework-tracked metadata contributed by a + // single caller request to this component. + message RequestMetadata { + // Links attached by this request. Components use these to record + // references back to the caller(s) that initiated or extended the + // execution (e.g. parent workflow events, Nexus completion sources). + repeated temporal.api.common.v1.Link links = 1; + } // Registered component's type ID. // (-- api-linter: core::0141::forbidden-types=disabled --) @@ -64,6 +73,11 @@ message ChasmComponentAttributes { // Detached components can continue operating, accepting writes and executing // tasks, even when their parent is closed/terminated. bool detached = 4; + // Per-request metadata contributed to this component, keyed by the caller's + // request_id. + map requests = 5; + // Caller-supplied user metadata (summary, details) attached to this component. + temporal.api.sdk.v1.UserMetadata user_metadata = 6; } message ChasmDataAttributes {}