Added the workflow publish and subscribe commands.

chasm/lib/stream/config.go

@@ -30,6 +30,12 @@ const LongPollBuffer = 3 * time.Second
 // shard would otherwise stretch the lock hold to match it.
 const RoutedCallTimeout = 5 * time.Second
 
+// RoutedSetBudget bounds every routed call one workflow task makes while the
+// execution's lock is held, taken together. RoutedCallTimeout bounds one of
+// them, and a task can carry as many as the subscription limit allows, so
+// without this the lock hold grows with that count.
+const RoutedSetBudget = 15 * time.Second
+
 // MaxListPageSize bounds a visibility page when the caller does not.
 const MaxListPageSize = 1000
 
@@ -96,6 +102,12 @@ const (
 	// that the byte budget alone does not see.
 	OwnedStreamMaxItems = 10_000
 
+	// MaxSubscriptionsPerWorkflow bounds how many streams one execution
+	// consumes. Each subscription costs a routed call on the completion path
+	// that made it and another on every task start, both with the execution's
+	// lock held, so the count is what bounds the lock hold.
+	MaxSubscriptionsPerWorkflow = 100
+
 	// OwnedStreamsMaxBytesPerWorkflow bounds every stream one execution owns
 	// taken together. The per-stream budget multiplied by the stream count
 	// comes to far more than limit.mutableStateSize.error, so without this an
@@ -163,6 +175,11 @@ under limit.mutableStateSize.error, which would otherwise terminate the workflow
 		OwnedStreamMaxItems,
 		`Message budget of a stream a workflow owns. Appends past it are refused.`,
 	)
+	MaxSubscriptionsPerWorkflowSetting = dynamicconfig.NewNamespaceIntSetting(
+		"stream.maxSubscriptionsPerWorkflow",
+		MaxSubscriptionsPerWorkflow,
+		`Most streams one workflow execution can consume.`,
+	)
 	OwnedStreamsMaxBytesPerWorkflowSetting = dynamicconfig.NewNamespaceIntSetting(
 		"stream.ownedStreamsMaxBytesPerWorkflow",
 		OwnedStreamsMaxBytesPerWorkflow,
@@ -197,6 +214,7 @@ type Config struct {
 	// Bounds every stream one execution owns taken together, which the
 	// per-stream budget cannot do.
 	OwnedStreamsMaxBytesPerWorkflow dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxSubscriptionsPerWorkflow     dynamicconfig.IntPropertyFnWithNamespaceFilter
 }
 
 func NewConfig(dc *dynamicconfig.Collection) *Config {
@@ -215,6 +233,7 @@ func NewConfig(dc *dynamicconfig.Collection) *Config {
 		OwnedStreamMaxItems:        OwnedStreamMaxItemsSetting.Get(dc),
 
 		OwnedStreamsMaxBytesPerWorkflow: OwnedStreamsMaxBytesPerWorkflowSetting.Get(dc),
+		MaxSubscriptionsPerWorkflow:     MaxSubscriptionsPerWorkflowSetting.Get(dc),
 	}
 }
 
@@ -232,6 +251,7 @@ type Limits struct {
 	OwnedStreamMaxItems        int
 
 	OwnedStreamsMaxBytesPerWorkflow int
+	MaxSubscriptionsPerWorkflow     int
 }
 
 // LimitsFor resolves the limits for a namespace. A nil Config, which is what
@@ -252,6 +272,7 @@ func (c *Config) LimitsFor(namespaceName string) Limits {
 		OwnedStreamMaxItems:        c.OwnedStreamMaxItems(namespaceName),
 
 		OwnedStreamsMaxBytesPerWorkflow: c.OwnedStreamsMaxBytesPerWorkflow(namespaceName),
+		MaxSubscriptionsPerWorkflow:     c.MaxSubscriptionsPerWorkflow(namespaceName),
 	}.withDefaults()
 }
 
@@ -296,5 +317,6 @@ func (l Limits) withDefaults() Limits {
 	fill(&l.OwnedStreamMaxBytes, OwnedStreamMaxBytes)
 	fill(&l.OwnedStreamMaxItems, OwnedStreamMaxItems)
 	fill(&l.OwnedStreamsMaxBytesPerWorkflow, OwnedStreamsMaxBytesPerWorkflow)
+	fill(&l.MaxSubscriptionsPerWorkflow, MaxSubscriptionsPerWorkflow)
 	return l
 }

chasm/lib/stream/service/events.go

@@ -0,0 +1,78 @@
+package service
+
+import (
+	enumspb "go.temporal.io/api/enums/v1"
+	historypb "go.temporal.io/api/history/v1"
+	"go.temporal.io/server/service/history/hsm"
+)
+
+// streamSubscribedEventDefinition tells the history service how to treat the
+// event a subscribe command writes.
+//
+// It applies nothing. The cursor lives in CHASM state, which is persisted and
+// rebuilt with the execution, so replication and reset have nothing to
+// reconstruct from the event. It exists so the command has an event at all:
+// every SDK matches issued commands against command-generated events in order,
+// and a command producing none puts that matching out of step.
+type streamSubscribedEventDefinition struct{}
+
+func (streamSubscribedEventDefinition) Type() enumspb.EventType {
+	return enumspb.EVENT_TYPE_WORKFLOW_STREAM_SUBSCRIBED
+}
+
+// Subscribing does not itself give the workflow anything to decide on. The
+// range that follows does, and that wakes the workflow through its cursor.
+func (streamSubscribedEventDefinition) IsWorkflowTaskTrigger() bool { return false }
+
+func (streamSubscribedEventDefinition) Apply(*hsm.Node, *historypb.HistoryEvent) error {
+	return nil
+}
+
+// A command event, so never reapplied onto another branch: a workflow that
+// still wants the subscription issues the command again on the new one.
+func (streamSubscribedEventDefinition) CherryPick(
+	*hsm.Node,
+	*historypb.HistoryEvent,
+	map[enumspb.ResetReapplyExcludeType]struct{},
+) error {
+	return hsm.ErrNotCherryPickable
+}
+
+// streamRecordsAppendedEventDefinition tells the history service how to treat
+// the event a publish command writes.
+//
+// Like the subscription event it applies nothing: the stream's frontier is
+// CHASM state committed with the workflow task, and the bodies are in the
+// stream component. What the event carries is the offset range, which is what
+// lets anyone reading History find the batch without History having held it.
+type streamRecordsAppendedEventDefinition struct{}
+
+func (streamRecordsAppendedEventDefinition) Type() enumspb.EventType {
+	return enumspb.EVENT_TYPE_WORKFLOW_STREAM_RECORDS_APPENDED
+}
+
+// A workflow publishing to its own stream has nothing to be woken about.
+func (streamRecordsAppendedEventDefinition) IsWorkflowTaskTrigger() bool { return false }
+
+func (streamRecordsAppendedEventDefinition) Apply(*hsm.Node, *historypb.HistoryEvent) error {
+	return nil
+}
+
+// A command event, so never reapplied onto another branch. Reapplying would
+// claim offsets in a log the new branch never wrote to.
+func (streamRecordsAppendedEventDefinition) CherryPick(
+	*hsm.Node,
+	*historypb.HistoryEvent,
+	map[enumspb.ResetReapplyExcludeType]struct{},
+) error {
+	return hsm.ErrNotCherryPickable
+}
+
+// RegisterEventDefinitions makes the stream events known to the history
+// service.
+func RegisterEventDefinitions(reg *hsm.Registry) error {
+	if err := reg.RegisterEventDefinition(streamSubscribedEventDefinition{}); err != nil {
+		return err
+	}
+	return reg.RegisterEventDefinition(streamRecordsAppendedEventDefinition{})
+}

chasm/lib/stream/service/fx.go

@@ -22,6 +22,7 @@ var HistoryModule = fx.Module(
 	fx.Invoke(func(l *library, registry *chasm.Registry) error {
 		return registry.Register(l)
 	}),
+	fx.Invoke(RegisterEventDefinitions),
 )
 
 var FrontendModule = fx.Module(

chasm/lib/workflow/fx.go

@@ -20,12 +20,16 @@ var Module = fx.Module(
 		chasmRegistry *chasm.Registry,
 		library *library,
 		config *nexusoperation.Config,
+		streamConfig *stream.Config,
 	) error {
 		if err := library.registry.Register(
 			newNexusLibrary(config, chasmRegistry.NexusEndpointProcessor),
 		); err != nil {
 			return err
 		}
+		if err := library.registry.Register(newStreamLibrary(streamConfig)); err != nil {
+			return err
+		}
 		return chasmRegistry.Register(library)
 	}),
 )

chasm/lib/workflow/stream_admission_test.go

@@ -4,7 +4,10 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	commandpb "go.temporal.io/api/command/v1"
 	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/serviceerror"
 	streampb "go.temporal.io/api/stream/v1"
 	"go.temporal.io/server/chasm"
@@ -114,3 +117,48 @@ func TestKnownHeadIsOnlyPushedIntoAnExternalCursor(t *testing.T) {
 	var notFound *serviceerror.NotFound
 	require.ErrorAs(t, err, &notFound)
 }
+
+// Each subscription costs a routed call on the completion path that made it,
+// with this execution's lock held, and another on every task start. Nothing
+// else bounds how many a workflow may hold or how many one task may carry.
+func TestSubscriptionsPerWorkflowAreBounded(t *testing.T) {
+	ctx := newStreamBudgetTestContext()
+	backend := &chasm.MockNodeBackend{
+		HandleAddHistoryEvent: func(
+			eventType enumspb.EventType, set func(*historypb.HistoryEvent),
+		) *historypb.HistoryEvent {
+			e := &historypb.HistoryEvent{EventType: eventType}
+			set(e)
+			return e
+		},
+	}
+	w := &Workflow{MSPointer: chasm.NewMSPointer(backend)}
+	opts := CommandHandlerOptions{WorkflowTaskCompletedEventID: 10}
+	limits := stream.Limits{MaxSubscriptionsPerWorkflow: 2}
+
+	subscribe := func(id string) error {
+		return handleSubscribeStreamCommand(ctx, w, nil, &commandpb.Command{
+			CommandType: enumspb.COMMAND_TYPE_SUBSCRIBE_STREAM,
+			Attributes: &commandpb.Command_SubscribeStreamCommandAttributes{
+				SubscribeStreamCommandAttributes: &commandpb.SubscribeStreamCommandAttributes{
+					StreamNameOrId: id,
+				},
+			},
+		}, opts, limits)
+	}
+
+	require.NoError(t, subscribe("a"))
+	require.NoError(t, subscribe("b"))
+
+	// Counted against what this task has already staged, so one task cannot
+	// carry an unbounded set of them either.
+	err := subscribe("c")
+	var failTask FailWorkflowTaskError
+	require.ErrorAs(t, err, &failTask)
+	require.Equal(t,
+		enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES, failTask.Cause)
+
+	// Subscribing again to one already staged registers nothing new, so it is
+	// not refused for room.
+	require.NoError(t, subscribe("a"))
+}