Added the stream service and its frontend wiring.

chasm/lib/stream/config.go

@@ -95,13 +95,32 @@ const (
 	// the same reason: each batch is a node, and many small ones cost state
 	// that the byte budget alone does not see.
 	OwnedStreamMaxItems = 10_000
+
+	// 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
+	// outside writer can name enough streams to terminate the execution while
+	// every single stream stays inside its own budget. Half the error limit,
+	// which leaves the rest of mutable state its own room.
+	OwnedStreamsMaxBytesPerWorkflow = 4 << 20
 )
 
 var (
+	// EnabledSetting gates the whole feature. Off by default: registering
+	// StreamService on the frontend otherwise turns a large new surface on in
+	// every deployment the moment it ships, and an operator needs a way to take
+	// it back without a rollback.
+	EnabledSetting = dynamicconfig.NewNamespaceBoolSetting(
+		"stream.enabled",
+		false,
+		`Whether the stream service and the workflow stream commands are available to a
+namespace. Off by default.`,
+	)
 	MaxConsumeItemsPerTaskSetting = dynamicconfig.NewNamespaceIntSetting(
 		"stream.maxConsumeItemsPerTask",
 		MaxConsumeItemsPerTask,
-		`Most stream records one workflow task carries per subscription.`,
+		`Most stream records one workflow task carries per subscription. Clamped to 1000,
+which is the largest page a single stream read returns.`,
 	)
 	MaxConsumeBytesPerTaskSetting = dynamicconfig.NewNamespaceIntSetting(
 		"stream.maxConsumeBytesPerTask",
@@ -144,6 +163,13 @@ under limit.mutableStateSize.error, which would otherwise terminate the workflow
 		OwnedStreamMaxItems,
 		`Message budget of a stream a workflow owns. Appends past it are refused.`,
 	)
+	OwnedStreamsMaxBytesPerWorkflowSetting = dynamicconfig.NewNamespaceIntSetting(
+		"stream.ownedStreamsMaxBytesPerWorkflow",
+		OwnedStreamsMaxBytesPerWorkflow,
+		`Byte budget of every stream one workflow execution owns, taken together. Appends
+past it are refused. Keep it under limit.mutableStateSize.error, which would otherwise
+terminate the workflow.`,
+	)
 	RetentionRecheckIntervalSetting = dynamicconfig.NewGlobalDurationSetting(
 		"stream.retentionRecheckInterval",
 		time.Minute,
@@ -154,6 +180,7 @@ consumers holding it are still running.`,
 
 // Config holds the settings as live property functions.
 type Config struct {
+	Enabled dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	// The id length limit shared with workflow ids. A stream id becomes an
 	// execution's business id, and a stream name a key in mutable state.
 	MaxIDLength                dynamicconfig.IntPropertyFn
@@ -167,10 +194,14 @@ type Config struct {
 	MaxOwnedStreamsPerWorkflow dynamicconfig.IntPropertyFnWithNamespaceFilter
 	OwnedStreamMaxBytes        dynamicconfig.IntPropertyFnWithNamespaceFilter
 	OwnedStreamMaxItems        dynamicconfig.IntPropertyFnWithNamespaceFilter
+	// Bounds every stream one execution owns taken together, which the
+	// per-stream budget cannot do.
+	OwnedStreamsMaxBytesPerWorkflow dynamicconfig.IntPropertyFnWithNamespaceFilter
 }
 
 func NewConfig(dc *dynamicconfig.Collection) *Config {
 	return &Config{
+		Enabled:                    EnabledSetting.Get(dc),
 		MaxIDLength:                dynamicconfig.MaxIDLengthLimit.Get(dc),
 		RetentionRecheckInterval:   RetentionRecheckIntervalSetting.Get(dc),
 		MaxConsumeItemsPerTask:     MaxConsumeItemsPerTaskSetting.Get(dc),
@@ -182,6 +213,8 @@ func NewConfig(dc *dynamicconfig.Collection) *Config {
 		MaxOwnedStreamsPerWorkflow: MaxOwnedStreamsPerWorkflowSetting.Get(dc),
 		OwnedStreamMaxBytes:        OwnedStreamMaxBytesSetting.Get(dc),
 		OwnedStreamMaxItems:        OwnedStreamMaxItemsSetting.Get(dc),
+
+		OwnedStreamsMaxBytesPerWorkflow: OwnedStreamsMaxBytesPerWorkflowSetting.Get(dc),
 	}
 }
 
@@ -197,6 +230,8 @@ type Limits struct {
 	MaxOwnedStreamsPerWorkflow int
 	OwnedStreamMaxBytes        int
 	OwnedStreamMaxItems        int
+
+	OwnedStreamsMaxBytesPerWorkflow int
 }
 
 // LimitsFor resolves the limits for a namespace. A nil Config, which is what
@@ -215,9 +250,21 @@ func (c *Config) LimitsFor(namespaceName string) Limits {
 		MaxOwnedStreamsPerWorkflow: c.MaxOwnedStreamsPerWorkflow(namespaceName),
 		OwnedStreamMaxBytes:        c.OwnedStreamMaxBytes(namespaceName),
 		OwnedStreamMaxItems:        c.OwnedStreamMaxItems(namespaceName),
+
+		OwnedStreamsMaxBytesPerWorkflow: c.OwnedStreamsMaxBytesPerWorkflow(namespaceName),
 	}.withDefaults()
 }
 
+// EnabledFor reports whether a namespace may use streams. A nil Config, which
+// is what component code driven without a service gets, reads as enabled: the
+// gate is a deployment switch, and a unit test is not a deployment.
+func (c *Config) EnabledFor(namespaceName string) bool {
+	if c == nil || c.Enabled == nil {
+		return true
+	}
+	return c.Enabled(namespaceName)
+}
+
 // DefaultLimits is the constant set above.
 func DefaultLimits() Limits {
 	return Limits{}.withDefaults()
@@ -236,6 +283,10 @@ func (l Limits) withDefaults() Limits {
 		}
 	}
 	fill(&l.MaxConsumeItemsPerTask, MaxConsumeItemsPerTask)
+	// A slice is built from one stream read, which serves at most a page, so a
+	// larger setting than that cannot take effect. Clamped here rather than
+	// left to disagree with what delivery does.
+	l.MaxConsumeItemsPerTask = min(l.MaxConsumeItemsPerTask, DefaultMaxMessagesPerPoll)
 	fill(&l.MaxConsumeBytesPerTask, MaxConsumeBytesPerTask)
 	fill(&l.MaxProducersPerStream, MaxProducersPerStream)
 	fill(&l.MaxConsumersPerStream, MaxConsumersPerStream)
@@ -244,5 +295,6 @@ func (l Limits) withDefaults() Limits {
 	fill(&l.MaxOwnedStreamsPerWorkflow, MaxOwnedStreamsPerWorkflow)
 	fill(&l.OwnedStreamMaxBytes, OwnedStreamMaxBytes)
 	fill(&l.OwnedStreamMaxItems, OwnedStreamMaxItems)
+	fill(&l.OwnedStreamsMaxBytesPerWorkflow, OwnedStreamsMaxBytesPerWorkflow)
 	return l
 }

chasm/lib/stream/gen/streampb/v1/namespace.go

@@ -0,0 +1,75 @@
+package streampb
+
+// Every RPC here carries its namespace inside `frontend_request` rather than at
+// the top level, because the top-level field is the resolved namespace id the
+// frontend fills in before routing.
+//
+// The server's interceptors find a request's namespace by asserting it to
+// `interceptor.NamespaceNameGetter`, which wants `GetNamespace() string` on the
+// request itself. Without these the assertion falls through to the id getter,
+// which at the frontend is still empty, so namespace rate limits, request
+// validation, the authorization target, redirection and the long-poll deadline
+// all resolve to the empty namespace and silently do nothing.
+//
+// The generator has no way to express "read it from this nested field", so the
+// methods are written here, next to the generated types they belong to.
+
+func (x *CreateStreamRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *AddMessagesRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *FinishWritingRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *SubscribeWorkflowRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *PollMessagesRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *DescribeStreamRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *PollWorkflowMessagesRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *DescribeWorkflowStreamRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *AddWorkflowMessagesRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *RegisterStreamConsumerRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *AdvanceConsumerHeadRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *CloseStreamRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *TruncateStreamRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *ListStreamsRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}
+
+func (x *DeleteStreamRequest) GetNamespace() string {
+	return x.GetFrontendRequest().GetNamespace()
+}

chasm/lib/stream/gen/streampb/v1/namespace_test.go

@@ -0,0 +1,38 @@
+package streampb
+
+import (
+	"testing"
+
+	"go.temporal.io/server/common/rpc/interceptor"
+	"google.golang.org/protobuf/reflect/protoregistry"
+)
+
+// Every request carrying a frontend_request must expose its namespace to the
+// interceptors. Driven off the descriptor rather than a hand-written list so a
+// new RPC fails here instead of silently opting out of namespace rate limits,
+// validation, authorization and redirection.
+func TestEveryRoutedRequestExposesNamespace(t *testing.T) {
+	fd := File_temporal_server_chasm_lib_stream_proto_v1_request_response_proto
+	messages := fd.Messages()
+
+	checked := 0
+	for i := 0; i < messages.Len(); i++ {
+		md := messages.Get(i)
+		if md.Fields().ByName("frontend_request") == nil {
+			continue
+		}
+		mt, err := protoregistry.GlobalTypes.FindMessageByName(md.FullName())
+		if err != nil {
+			t.Fatalf("%s is not registered: %v", md.FullName(), err)
+		}
+		msg := mt.New().Interface()
+		if _, ok := msg.(interceptor.NamespaceNameGetter); !ok {
+			t.Errorf("%s has a frontend_request but no GetNamespace; add it in namespace.go", md.FullName())
+		}
+		checked++
+	}
+
+	if checked == 0 {
+		t.Fatal("found no routed requests, so this test is not checking anything")
+	}
+}