[Jetty] Closed a Workflow Stream from a System Nexus command of its chain.

chasm/lib/activity/activity_test.go

@@ -1149,8 +1149,8 @@ func TestEffectiveUserMetadata_FallsBackToLegacy(t *testing.T) {
 	require.Same(t, legacyMD, got)
 }
 
-// A Stream ref with an activity id routes to the Activity's component, so the stream library's
-// copy of the archetype must stay this library's.
+// A Stream ref with an activity id routes to stream.ActivityArchetypeID, so that id must equal
+// this library's ArchetypeID.
 func TestStreamHostArchetypeIsTheActivity(t *testing.T) {
 	require.Equal(t, ArchetypeID, stream.ActivityArchetypeID)
 }

chasm/lib/activity/stream.go

@@ -8,9 +8,8 @@ import (
 	"go.temporal.io/server/common/tasktoken"
 )
 
-// A Standalone Activity owns its Streams. They nest in the Activity's execution, so a call routes
-// by the activity id, and they close in the transition that ends the Activity, with its status.
-// A retry is the same execution, so a Stream goes on across attempts.
+// A Standalone Activity's Streams nest in its execution, so a call routes by the activity id. A
+// retry is the same execution, so a Stream stays open across attempts.
 
 var _ stream.Host = (*Activity)(nil)
 
@@ -22,8 +21,8 @@ func (a *Activity) NestedStreams() chasm.Map[string, *stream.Stream] {
 	return a.Streams
 }
 
-// CheckOwner refuses a create that isn't the Activity's own, by its current attempt. It
-// implements [stream.Host].
+// CheckOwner refuses a create unless the owner is the Activity's own Producer with the task token
+// of its current attempt. It implements [stream.Host].
 func (a *Activity) CheckOwner(
 	ctx chasm.Context,
 	owner *streamservice.Producer,
@@ -35,8 +34,8 @@ func (a *Activity) CheckOwner(
 	return a.checkAttempt(ctx, owner, token)
 }
 
-// CheckWriter refuses a write as the Activity's own Producer by any other than its current
-// attempt. Other Producers write as on any Stream. It implements [stream.Host].
+// CheckWriter refuses a write as the Activity's own Producer unless the task token is the
+// Activity's current attempt. It doesn't check other Producers. It implements [stream.Host].
 func (a *Activity) CheckWriter(
 	ctx chasm.Context,
 	producer *streamservice.Producer,
@@ -48,7 +47,7 @@ func (a *Activity) CheckWriter(
 	return a.checkAttempt(ctx, producer, token)
 }
 
-// ownProducer says whether a Producer is this Activity's, by the id that the crate makes.
+// ownProducer says whether a Producer has the kind and the id that the crate gives this Activity.
 func (a *Activity) ownProducer(ctx chasm.Context, producer *streamservice.Producer) bool {
 	key := ctx.ExecutionKey()
 	id := "sa:" + stream.EncodeIDPart(key.BusinessID) + ":" + stream.EncodeIDPart(key.RunID)
@@ -57,7 +56,7 @@ func (a *Activity) ownProducer(ctx chasm.Context, producer *streamservice.Produc
 }
 
 // checkAttempt checks that a task token is this Activity's current attempt, and that the
-// Producer's epoch is that attempt. The epoch stays (attempt scheduled time, attempt).
+// Producer's epoch has the same attempt.
 func (a *Activity) checkAttempt(
 	ctx chasm.Context,
 	producer *streamservice.Producer,
@@ -81,7 +80,13 @@ func (a *Activity) checkAttempt(
 	return nil
 }
 
-// endStreams closes the Activity's Streams with its end. Each terminal transition passes its
+// HoldsStreams says whether a Stream of the closed Activity is still readable, so the Activity
+// must outlive the namespace retention. A closed Stream stays readable for its Retention.
+func (a *Activity) HoldsStreams(ctx chasm.Context) bool {
+	return stream.HoldsStreams(ctx, a.Streams)
+}
+
+// endStreams closes the Activity's Streams when it ends. Each terminal transition passes its
 // status, since the transition sets the Activity's status after its apply function.
 func (a *Activity) endStreams(ctx chasm.MutableContext, status activitypb.ActivityExecutionStatus) {
 	if len(a.Streams) == 0 {

chasm/lib/stream/config.go

@@ -31,7 +31,8 @@ Stream that a client or an Activity created lives one Retention after its last w
 		"stream.maxStoredBytes",
 		4*1024*1024,
 		`The most Record bytes a Stream keeps. The Records are in the Stream's mutable state, so an
-append past the cap drops the oldest batches, as the Retention does.`,
+append past the cap drops the oldest batches, as the Retention does. The append fails when an open
+Workflow reader still needs them.`,
 	)
 
 	MaxWorkflowStateBytes = dynamicconfig.NewNamespaceIntSetting(
@@ -57,15 +58,15 @@ retryable error, so the caller backs off.`,
 	MaxWorkflowReaders = dynamicconfig.NewNamespaceIntSetting(
 		"stream.maxWorkflowReaders",
 		100,
-		`The most open Workflow runs whose read floor one Workflow Stream keeps. A page of a new
-reader past the cap fails with a retryable error, so the caller backs off.`,
+		`The most open Workflow runs whose read floor one Stream keeps. A page of a new reader past
+the cap fails with a retryable error, so the caller backs off.`,
 	)
 
 	ReaderCheckInterval = dynamicconfig.NewNamespaceDurationSetting(
 		"stream.readerCheckInterval",
 		5*time.Minute,
-		`How often a Workflow Stream with read floors checks whether their runs are still open. The
-floor of a closed run goes at the next check.`,
+		`How often a Stream with read floors checks whether their runs are still open. The floor of
+a closed run goes at the next check.`,
 	)
 
 	MaxWakeWait = dynamicconfig.NewNamespaceDurationSetting(

chasm/lib/stream/errors.go

@@ -48,7 +48,8 @@ var reasons = []string{
 // StreamErrorDetails in the status details.
 func NewError(reason string, format string, args ...any) error {
 	st := status.New(reasonCode(reason), reason+": "+fmt.Sprintf(format, args...))
-	if detailed, err := st.WithDetails(&streamservice.StreamErrorDetails{Reason: reason}); err == nil {
+	detailed, err := st.WithDetails(&streamservice.StreamErrorDetails{Reason: reason})
+	if err == nil {
 		st = detailed
 	}
 	return serviceerror.FromStatus(st)

chasm/lib/stream/frontend.go

@@ -28,16 +28,26 @@ func NewFrontendHandler(
 	return &FrontendHandler{client: client, config: config, namespaceRegistry: namespaceRegistry}
 }
 
-// RedirectableMethods lists the StreamService methods that a cell forwards to the namespace's
-// active cell, with the response each one answers. The Stream is in the active cell's
-// mutable state, so a call served elsewhere would read a replica or write a copy.
+// RedirectableMethods maps each StreamService method that a cell forwards to the namespace's
+// active cell to a func that makes its empty response. The Stream is in the active cell's mutable
+// state, so a call served elsewhere would read a replica or write a copy.
 func RedirectableMethods() map[string]func() any {
 	return map[string]func() any{
-		streamservice.StreamService_CreateStream_FullMethodName:   func() any { return &streamservice.CreateStreamResponse{} },
-		streamservice.StreamService_AppendRecords_FullMethodName:  func() any { return &streamservice.AppendRecordsResponse{} },
-		streamservice.StreamService_CloseStream_FullMethodName:    func() any { return &streamservice.CloseStreamResponse{} },
-		streamservice.StreamService_ReadRecords_FullMethodName:    func() any { return &streamservice.ReadRecordsResponse{} },
-		streamservice.StreamService_DescribeStream_FullMethodName: func() any { return &streamservice.DescribeStreamResponse{} },
+		streamservice.StreamService_CreateStream_FullMethodName: func() any {
+			return &streamservice.CreateStreamResponse{}
+		},
+		streamservice.StreamService_AppendRecords_FullMethodName: func() any {
+			return &streamservice.AppendRecordsResponse{}
+		},
+		streamservice.StreamService_CloseStream_FullMethodName: func() any {
+			return &streamservice.CloseStreamResponse{}
+		},
+		streamservice.StreamService_ReadRecords_FullMethodName: func() any {
+			return &streamservice.ReadRecordsResponse{}
+		},
+		streamservice.StreamService_DescribeStream_FullMethodName: func() any {
+			return &streamservice.DescribeStreamResponse{}
+		},
 	}
 }
 
@@ -70,7 +80,10 @@ func (h *FrontendHandler) CreateStream(
 	if err != nil {
 		return nil, err
 	}
-	resp, err := h.client.CreateStream(ctx, &streampb.CreateStreamRequest{NamespaceId: namespaceID, FrontendRequest: req})
+	resp, err := h.client.CreateStream(
+		ctx,
+		&streampb.CreateStreamRequest{NamespaceId: namespaceID, FrontendRequest: req},
+	)
 	return resp.GetFrontendResponse(), err
 }
 
@@ -82,7 +95,10 @@ func (h *FrontendHandler) AppendRecords(
 	if err != nil {
 		return nil, err
 	}
-	resp, err := h.client.AppendRecords(ctx, &streampb.AppendRecordsRequest{NamespaceId: namespaceID, FrontendRequest: req})
+	resp, err := h.client.AppendRecords(
+		ctx,
+		&streampb.AppendRecordsRequest{NamespaceId: namespaceID, FrontendRequest: req},
+	)
 	return resp.GetFrontendResponse(), err
 }
 
@@ -94,7 +110,10 @@ func (h *FrontendHandler) CloseStream(
 	if err != nil {
 		return nil, err
 	}
-	resp, err := h.client.CloseStream(ctx, &streampb.CloseStreamRequest{NamespaceId: namespaceID, FrontendRequest: req})
+	resp, err := h.client.CloseStream(
+		ctx,
+		&streampb.CloseStreamRequest{NamespaceId: namespaceID, FrontendRequest: req},
+	)
 	return resp.GetFrontendResponse(), err
 }
 
@@ -106,7 +125,10 @@ func (h *FrontendHandler) ReadRecords(
 	if err != nil {
 		return nil, err
 	}
-	resp, err := h.client.ReadRecords(ctx, &streampb.ReadRecordsRequest{NamespaceId: namespaceID, FrontendRequest: req})
+	resp, err := h.client.ReadRecords(
+		ctx,
+		&streampb.ReadRecordsRequest{NamespaceId: namespaceID, FrontendRequest: req},
+	)
 	return resp.GetFrontendResponse(), err
 }
 
@@ -118,6 +140,9 @@ func (h *FrontendHandler) DescribeStream(
 	if err != nil {
 		return nil, err
 	}
-	resp, err := h.client.DescribeStream(ctx, &streampb.DescribeStreamRequest{NamespaceId: namespaceID, FrontendRequest: req})
+	resp, err := h.client.DescribeStream(
+		ctx,
+		&streampb.DescribeStreamRequest{NamespaceId: namespaceID, FrontendRequest: req},
+	)
 	return resp.GetFrontendResponse(), err
 }

chasm/lib/stream/grpc.go

@@ -17,39 +17,54 @@ func newGRPCHandler(h *handler) *grpcHandler {
 	return &grpcHandler{handler: h}
 }
 
-func (g *grpcHandler) CreateStream(ctx context.Context, req *streampb.CreateStreamRequest) (*streampb.CreateStreamResponse, error) {
+func (g *grpcHandler) CreateStream(
+	ctx context.Context,
+	req *streampb.CreateStreamRequest,
+) (*streampb.CreateStreamResponse, error) {
 	resp, err := g.handler.CreateStream(ctx, req.GetFrontendRequest())
 	if err != nil {
 		return nil, err
 	}
 	return &streampb.CreateStreamResponse{FrontendResponse: resp}, nil
 }
 
-func (g *grpcHandler) AppendRecords(ctx context.Context, req *streampb.AppendRecordsRequest) (*streampb.AppendRecordsResponse, error) {
+func (g *grpcHandler) AppendRecords(
+	ctx context.Context,
+	req *streampb.AppendRecordsRequest,
+) (*streampb.AppendRecordsResponse, error) {
 	resp, err := g.handler.AppendRecords(ctx, req.GetFrontendRequest())
 	if err != nil {
 		return nil, err
 	}
 	return &streampb.AppendRecordsResponse{FrontendResponse: resp}, nil
 }
 
-func (g *grpcHandler) CloseStream(ctx context.Context, req *streampb.CloseStreamRequest) (*streampb.CloseStreamResponse, error) {
+func (g *grpcHandler) CloseStream(
+	ctx context.Context,
+	req *streampb.CloseStreamRequest,
+) (*streampb.CloseStreamResponse, error) {
 	resp, err := g.handler.CloseStream(ctx, req.GetFrontendRequest())
 	if err != nil {
 		return nil, err
 	}
 	return &streampb.CloseStreamResponse{FrontendResponse: resp}, nil
 }
 
-func (g *grpcHandler) ReadRecords(ctx context.Context, req *streampb.ReadRecordsRequest) (*streampb.ReadRecordsResponse, error) {
+func (g *grpcHandler) ReadRecords(
+	ctx context.Context,
+	req *streampb.ReadRecordsRequest,
+) (*streampb.ReadRecordsResponse, error) {
 	resp, err := g.handler.ReadRecords(ctx, req.GetFrontendRequest(), nil)
 	if err != nil {
 		return nil, err
 	}
 	return &streampb.ReadRecordsResponse{FrontendResponse: resp}, nil
 }
 
-func (g *grpcHandler) DescribeStream(ctx context.Context, req *streampb.DescribeStreamRequest) (*streampb.DescribeStreamResponse, error) {
+func (g *grpcHandler) DescribeStream(
+	ctx context.Context,
+	req *streampb.DescribeStreamRequest,
+) (*streampb.DescribeStreamResponse, error) {
 	resp, err := g.handler.DescribeStream(ctx, req.GetFrontendRequest())
 	if err != nil {
 		return nil, err