[CHASM] Wire ChasmRegistry into shard.Context (temporalio#7419)

## What changed?
- Wires up CHASM registry to shard context

## Why?
- It's needed throughout history service 

## How did you test it?


## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->

## Documentation
<!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? -->

## Is hotfix candidate?
<!-- Is this PR a hotfix candidate or does it require a notification to
be sent to the broader community? (Yes/No) -->

Author: lina-temporal

service/history/fx.go

@@ -26,6 +26,7 @@ package history
 
 import (
 	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/config"
@@ -70,6 +71,7 @@ import (
 var Module = fx.Options(
 	resource.Module,
 	fx.Provide(hsm.NewRegistry),
+	fx.Provide(chasm.NewRegistry),
 	workflow.Module,
 	shard.Module,
 	events.Module,

service/history/shard/context.go

@@ -33,6 +33,7 @@ import (
 	clockspb "go.temporal.io/server/api/clock/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/common/archiver"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/cluster"
@@ -123,6 +124,8 @@ type (
 
 		StateMachineRegistry() *hsm.Registry
 		GetFinalizer() *finalizer.Finalizer
+
+		ChasmRegistry() *chasm.Registry
 	}
 
 	// A ControllableContext is a Context plus other methods needed by

service/history/shard/context_factory.go

@@ -25,6 +25,7 @@
 package shard
 
 import (
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/client"
 	"go.temporal.io/server/common/archiver"
 	"go.temporal.io/server/common/clock"
@@ -80,6 +81,7 @@ type (
 		EventsCache                 events.Cache
 
 		StateMachineRegistry *hsm.Registry
+		ChasmRegistry        *chasm.Registry
 	}
 
 	contextFactoryImpl struct {
@@ -121,6 +123,7 @@ func (c *contextFactoryImpl) CreateContext(
 		c.TaskCategoryRegistry,
 		c.EventsCache,
 		c.StateMachineRegistry,
+		c.ChasmRegistry,
 	)
 	if err != nil {
 		return nil, err

service/history/shard/context_impl.go

@@ -43,6 +43,7 @@ import (
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/client"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/archiver"
@@ -169,6 +170,8 @@ type (
 		acquireShardRetryPolicy backoff.RetryPolicy
 
 		stateMachineRegistry *hsm.Registry
+
+		chasmRegistry *chasm.Registry
 	}
 
 	remoteClusterInfo struct {
@@ -2088,6 +2091,7 @@ func newContext(
 	taskCategoryRegistry tasks.TaskCategoryRegistry,
 	eventsCache events.Cache,
 	stateMachineRegistry *hsm.Registry,
+	chasmRegistry *chasm.Registry,
 ) (*ContextImpl, error) {
 	hostIdentity := hostInfoProvider.HostInfo().Identity()
 	sequenceID := atomic.AddInt64(&shardContextSequenceID, 1)
@@ -2136,6 +2140,7 @@ func newContext(
 		queueMetricEmitter:      sync.Once{},
 		ioSemaphore:             locks.NewPrioritySemaphore(ioConcurrency),
 		stateMachineRegistry:    stateMachineRegistry,
+		chasmRegistry:           chasmRegistry,
 	}
 	shardContext.taskKeyManager = newTaskKeyManager(
 		shardContext.taskCategoryRegistry,
@@ -2238,6 +2243,10 @@ func (s *ContextImpl) StateMachineRegistry() *hsm.Registry {
 	return s.stateMachineRegistry
 }
 
+func (s *ContextImpl) ChasmRegistry() *chasm.Registry {
+	return s.chasmRegistry
+}
+
 // newDetachedContext creates a detached context with the same deadline
 // and values from the given context. Detached context won't be affected
 // if the context it bases on is cancelled.

service/history/shard/context_mock.go

@@ -31,6 +31,7 @@ import (
 
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/future"
@@ -171,6 +172,7 @@ func newTestContext(t *resourcetest.Test, eventsCache events.Cache, config Conte
 		timeSource:              t.TimeSource,
 		namespaceRegistry:       registry,
 		stateMachineRegistry:    hsm.NewRegistry(),
+		chasmRegistry:           chasm.NewRegistry(),
 		persistenceShardManager: t.GetShardManager(),
 		clientBean:              t.GetClientBean(),
 		saProvider:              t.GetSearchAttributesProvider(),
@@ -227,6 +229,10 @@ func (s *ContextTest) SetStateMachineRegistry(reg *hsm.Registry) {
 	s.stateMachineRegistry = reg
 }
 
+func (s *ContextTest) SetChasmRegistry(reg *chasm.Registry) {
+	s.chasmRegistry = reg
+}
+
 // StopForTest calls FinishStop(). In general only the controller
 // should call that, but integration tests need to do it also to clean up any
 // background acquireShard goroutines that may exist.