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,
@@ -235,6 +252,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
 
 	AppendRecordsPerSecond dynamicconfig.IntPropertyFnWithNamespaceFilter
 	AppendBytesPerSecond   dynamicconfig.IntPropertyFnWithNamespaceFilter
@@ -258,6 +276,7 @@ func NewConfig(dc *dynamicconfig.Collection) *Config {
 		OwnedStreamMaxItems:        OwnedStreamMaxItemsSetting.Get(dc),
 
 		OwnedStreamsMaxBytesPerWorkflow: OwnedStreamsMaxBytesPerWorkflowSetting.Get(dc),
+		MaxSubscriptionsPerWorkflow:     MaxSubscriptionsPerWorkflowSetting.Get(dc),
 
 		AppendRecordsPerSecond: AppendRecordsPerSecondSetting.Get(dc),
 		AppendBytesPerSecond:   AppendBytesPerSecondSetting.Get(dc),
@@ -279,6 +298,7 @@ type Limits struct {
 	OwnedStreamMaxItems        int
 
 	OwnedStreamsMaxBytesPerWorkflow int
+	MaxSubscriptionsPerWorkflow     int
 }
 
 // LimitsFor resolves the limits for a namespace. A nil Config, which is what
@@ -299,6 +319,7 @@ func (c *Config) LimitsFor(namespaceName string) Limits {
 		OwnedStreamMaxItems:        c.OwnedStreamMaxItems(namespaceName),
 
 		OwnedStreamsMaxBytesPerWorkflow: c.OwnedStreamsMaxBytesPerWorkflow(namespaceName),
+		MaxSubscriptionsPerWorkflow:     c.MaxSubscriptionsPerWorkflow(namespaceName),
 	}.withDefaults()
 }
 
@@ -343,5 +364,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

@@ -23,6 +23,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"
@@ -116,6 +119,51 @@ func TestKnownHeadIsOnlyPushedIntoAnExternalCursor(t *testing.T) {
 	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"))
+}
+
 // An activity id may contain a slash, so the reserved key escapes it. Without
 // that two different (activity, name) pairs would land on one stream.
 func TestActivityStreamKeysDoNotCollide(t *testing.T) {
@@ -147,3 +195,38 @@ func TestWorkflowCannotNameAnActivityStream(t *testing.T) {
 		w.OwnedStream(ctx, key).State.GetAppendedBytes())
 	require.Len(t, w.Streams, 1)
 }
+
+// An activity's streams end when it reaches a terminal status, while the
+// workflow's own streams and another activity's stay open.
+func TestCloseActivityStreamsEndsOnlyThatActivity(t *testing.T) {
+	ctx := newStreamBudgetTestContext()
+	w := &Workflow{}
+	limits := stream.Limits{MaxOwnedStreamsPerWorkflow: 10}
+	for _, key := range []string{
+		DefaultStreamName,
+		ActivityStreamKey("act", DefaultStreamName),
+		ActivityStreamKey("act", "reasoning"),
+		ActivityStreamKey("act-2", DefaultStreamName),
+	} {
+		_, err := w.AppendToOwnedStream(ctx, key, stream.AddMessagesRequest{
+			Records: budgetTestRecords(1), Limits: limits,
+		})
+		require.NoError(t, err)
+	}
+
+	require.True(t, w.HasOpenActivityStreams(ctx, "act"))
+	require.False(t, w.HasOpenActivityStreams(ctx, "never-wrote"))
+	require.NoError(t, w.CloseActivityStreams(ctx, "act"))
+	require.False(t, w.HasOpenActivityStreams(ctx, "act"))
+
+	closed := func(key string) bool { return w.OwnedStream(ctx, key).State.GetClosed() }
+	require.True(t, closed(ActivityStreamKey("act", DefaultStreamName)))
+	require.True(t, closed(ActivityStreamKey("act", "reasoning")))
+	require.False(t, closed(ActivityStreamKey("act-2", DefaultStreamName)))
+	require.False(t, closed(DefaultStreamName))
+
+	_, err := w.AppendToOwnedStream(ctx, ActivityStreamKey("act", DefaultStreamName),
+		stream.AddMessagesRequest{Records: budgetTestRecords(1), Limits: limits})
+	var precondition *serviceerror.FailedPrecondition
+	require.ErrorAs(t, err, &precondition, "an ended activity's stream takes no more records")
+}