Files
tasks/chasm/ref.go
jiechenzandGitHub 929412e504 Route delete execution replication task by archetype ID (#9831)
## What changed?
On the passive clusters side, route delete execution replication tasks
based on `archetypeID` from `RawTaskInfo`:
- `WorkflowArchetypeID` or `UnspecifiedArchetypeID` →
`engine.DeleteWorkflowExecution` path
- Any other archetype → `ChasmEngine.DeleteExecution`

## Why?
Today, the sender (`context_impl.go`) only generates
`DeleteExecutionReplicationTask` for `WorkflowArchetypeID`. The receiver
previously dropped all non-workflow archetypes.

This change makes the receiver forward-compatible: when delete
replication is enabled for non-workflow Chasm archetypes (activity,
scheduler, etc.) in the future, the standby side will already handle
them via
`ChasmEngine.DeleteExecution`.

`UnspecifiedArchetypeID` (0) routes to the workflow path for backward
compatibility. Assumption is old senders don't populate `ArchetypeId` in
`RawTaskInfo`, and historically only workflows produced delete
replication tasks. Routing `0` to the Chasm path would fail because
`getExecutionLease` looks up `current_chasm_executions` by archetype ID.

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [x] covered by existing tests
- [x] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks

**`deleteChasmExecution` path is untested end-to-end**: The sender guard
in `context_impl.go:1032` prevents non-workflow delete replication tasks
from being generated, so the `ChasmEngine.DeleteExecution` branch cannot
be coverred in integration tests today
2026-04-08 15:50:38 -07:00

157 lines
5.2 KiB
Go

package chasm
import (
"reflect"
"go.temporal.io/api/serviceerror"
persistencespb "go.temporal.io/server/api/persistence/v1"
)
// ErrMalformedComponentRef is returned when component ref bytes cannot be deserialized.
var ErrMalformedComponentRef = serviceerror.NewInvalidArgument("malformed component ref")
// ErrInvalidComponentRef is returned when component ref bytes deserialize to an invalid component ref.
var ErrInvalidComponentRef = serviceerror.NewInvalidArgument("invalid component ref")
// ExecutionKey uniquely identifies a CHASM execution in the system.
type ExecutionKey struct {
NamespaceID string
BusinessID string
RunID string
}
type ComponentRef struct {
ExecutionKey
// archetypeID is CHASM framework's internal ID for the type of the root component of the CHASM execution.
//
// It is used to find and validate the loaded execution has the right archetype, especially when runID
// is not specified in the ExecutionKey.
archetypeID ArchetypeID
// executionGoType is used for determining the ComponetRef's archetype.
// When CHASM deverloper needs to create a ComponentRef, they will only provide the component type,
// and leave the work of determining archetypeID to the CHASM framework.
executionGoType reflect.Type
// executionLastUpdateVT is the consistency token for the entire execution.
executionLastUpdateVT *persistencespb.VersionedTransition
// componentType is the fully qualified component type name.
// It is for performing partial loading more efficiently in future versions of CHASM.
//
// From the componentType, we can find the registered component struct definition,
// then use reflection to find sub-components and understand if those sub-components
// need to be loaded or not.
// We only need to do this for sub-components, path for parent/ancenstor components
// can be inferred from the current component path and they always needs to be loaded.
//
// componentType string
// componentPath and componentInitialVT are used to identify a component.
componentPath []string
componentInitialVT *persistencespb.VersionedTransition
validationFn func(NodeBackend, Context, Component, *Registry) error
}
// NewComponentRef creates a new ComponentRef with a registered root component go type.
//
// In V1, if you don't have a ref,
// then you can only interact with the (top level) execution.
func NewComponentRef[C Component](
executionKey ExecutionKey,
) ComponentRef {
return ComponentRef{
ExecutionKey: executionKey,
executionGoType: reflect.TypeFor[C](),
}
}
// NewComponentRefByArchetypeID creates a new ComponentRef with a known archetype ID.
// This should only be used by CHASM framework internals.
// CHASM library developers should use [NewComponentRef] instead.
func NewComponentRefByArchetypeID(
executionKey ExecutionKey,
archetypeID ArchetypeID,
) ComponentRef {
return ComponentRef{
ExecutionKey: executionKey,
archetypeID: archetypeID,
}
}
func (r *ComponentRef) ArchetypeID(
registry *Registry,
) (ArchetypeID, error) {
if r.archetypeID != UnspecifiedArchetypeID {
return r.archetypeID, nil
}
rc, ok := registry.componentOf(r.executionGoType)
if !ok {
return 0, serviceerror.NewInternal("unknown chasm component type: " + r.executionGoType.String())
}
r.archetypeID = rc.componentID
return r.archetypeID, nil
}
func (r *ComponentRef) Serialize(
registry *Registry,
) ([]byte, error) {
if r == nil {
return nil, nil
}
archetypeID, err := r.ArchetypeID(registry)
if err != nil {
return nil, err
}
pRef := persistencespb.ChasmComponentRef{
NamespaceId: r.NamespaceID,
BusinessId: r.BusinessID,
RunId: r.RunID,
ArchetypeId: archetypeID,
ExecutionVersionedTransition: r.executionLastUpdateVT,
ComponentPath: r.componentPath,
ComponentInitialVersionedTransition: r.componentInitialVT,
}
return pRef.Marshal()
}
// DeserializeComponentRef deserializes a byte slice into a ComponentRef.
// Provides caller the access to information including ExecutionKey, Archetype, and ShardingKey.
func DeserializeComponentRef(data []byte) (ComponentRef, error) {
if len(data) == 0 {
return ComponentRef{}, ErrInvalidComponentRef
}
var pRef persistencespb.ChasmComponentRef
if err := pRef.Unmarshal(data); err != nil {
return ComponentRef{}, ErrMalformedComponentRef
}
ref := ProtoRefToComponentRef(&pRef)
if ref.BusinessID == "" || ref.NamespaceID == "" {
return ComponentRef{}, ErrInvalidComponentRef
}
return ref, nil
}
// ProtoRefToComponentRef converts a persistence ChasmComponentRef reference to a
// ComponentRef. This is useful for situations where the protobuf ComponentRef has
// already been deserialized as part of an enclosing message.
func ProtoRefToComponentRef(pRef *persistencespb.ChasmComponentRef) ComponentRef {
return ComponentRef{
ExecutionKey: ExecutionKey{
NamespaceID: pRef.NamespaceId,
BusinessID: pRef.BusinessId,
RunID: pRef.RunId,
},
archetypeID: pRef.ArchetypeId,
executionLastUpdateVT: pRef.ExecutionVersionedTransition,
componentPath: pRef.ComponentPath,
componentInitialVT: pRef.ComponentInitialVersionedTransition,
}
}