Added the temporal stream command group.

docs/stream.md

@@ -0,0 +1,145 @@
+# `temporal stream`
+
+Streams are append-only logs the Temporal Service carries next to Workflow
+Executions. This fork adds a `temporal stream` command group on top of the
+stream service, and a dev server that serves it.
+
+## Dev server
+
+`temporal server start-dev` registers the stream service and turns it on for
+every namespace it creates. The server ships with `stream.enabled` off, so the
+dev server sets it as one of its own dynamic-config defaults, the same way it
+sets the CHASM and standalone-activity flags. An explicit value still wins:
+
+```sh
+temporal server start-dev --dynamic-config-value 'stream.enabled=false'
+```
+
+The server behind the dev server is pinned in `go.mod` to the stream branch of
+`moedash/temporal`, with the matching `moedash/api-go` pin the server needs.
+
+## Addressing a stream
+
+Two kinds of stream exist. A standalone stream has an ID of its own. An owned
+stream lives inside a Workflow Execution or an Activity and is named by its
+owner and a stream name.
+
+- `--stream-id ID`: the standalone stream `ID`.
+- `--workflow-id W [--run-id R] [--name N]`: the stream `N` the workflow owns.
+  Without `--name` it is the workflow's default stream.
+- `--workflow-id W --activity-id A [--name N]`: the stream of an activity the
+  workflow scheduled.
+- `--activity-id A [--run-id R] [--name N]`: the stream of a standalone
+  activity.
+
+`describe`, `read` and `append` take either form. `create`, `close`,
+`truncate` and `delete` act on a standalone stream only, because an owned
+stream's lifecycle is its owner's.
+
+## Commands
+
+`stream create --stream-id ID [--retention D] [--max-items N] [--max-bytes N]`
+: Creates a standalone stream. Retention defaults to the namespace's; records
+  older than it are reclaimed. `--max-items` is a rolling window on records,
+  `--max-bytes` a ceiling on held bytes that refuses appends until records
+  are reclaimed. Repeating a create with the same lifecycle answers "already
+  exists"; a different lifecycle is refused.
+
+`stream list [--query Q] [--limit N] [--page-size N]`
+: Lists standalone streams from visibility. `WorkflowId` in the query is the
+  stream ID. Owned streams are not listed; reach them through their owner.
+
+`stream describe <ref>`
+: Frontier, floor, readable record count, held and appended bytes, close
+  state and reason, retention, caps, budget, producers and consumers. Also
+  the notification channel the stream notifies on each append and close,
+  derived from the reference alone: `stream/NAME` linked to the owner for a
+  workflow's or a standalone activity's stream, `stream/ACTIVITY_ID/NAME`
+  linked to the workflow for an activity it scheduled, and the independent
+  `stream/STREAM_ID` for a standalone stream. The card names the owner as
+  `LinkedTo workflow ID` or `LinkedTo activity ID`, with the run when one was
+  given. `temporal channel poll --channel C` with the same `--workflow-id` or
+  `--activity-id` follows it.
+
+`stream read <ref> [--from-offset N | --from-tail | --last N] [--follow]
+[--topic T]... [--limit N]`
+: Prints records with offset, kind, topic, producer, attempt, sequence and
+  body. Stops once caught up, or with `--follow` when the stream closes.
+
+`stream append <ref> --input V... [--topic T] [--producer-id P] [--attempt A]
+[--sequence S] [--expected-offset N] [--finish]`
+: Appends one record per `--input` or `--input-file`, or per line of stdin.
+  `--producer-id` with `--sequence` deduplicates a retry. `--finish` adds a
+  `FINISH` record after the values.
+
+`stream close --stream-id ID [--reason R]`
+: Seals the stream. Records stay readable until retention passes.
+
+`stream truncate --stream-id ID --to N`
+: Moves the floor to `N`. Refused below an active consumer's floor.
+
+`stream delete --stream-id ID [--force]`
+: Deletes the stream and its records. Refused while a workflow consumes it,
+  unless forced.
+
+`--output json` prints one JSON object per record or list entry, and the
+stream state proto for `describe`.
+
+Authorization follows the server's declaration: `truncate` and `delete` need
+the admin role on the namespace, `list`, `describe` and `read` the read role,
+and the rest the write role.
+
+## Notification channels
+
+`temporal channel notify|describe|poll|listener add|listener remove` reach the
+channel a stream notifies, or any other channel, by `--channel` (`-c`). Without
+an owner the commands use the independent channel of that name. With
+`--workflow-id W` or `--activity-id A`, and an optional `--run-id`, they use
+the channel linked to that Workflow Execution or standalone Activity, sent as
+the request's `execution` with the matching type. The two owner flags cannot
+be combined, and `--run-id` alone is refused.
+
+`channel describe` prints the kind and, for a linked channel, the owner line
+`LinkedTo workflow W (run R)` or `LinkedTo activity A (run R)`, the run only
+when the Service names one. The JSON output carries `linkedTo` as the Service
+sends it.
+
+## Refusals
+
+A refusal the caller has to act on comes back from the server as a
+`FailedPrecondition` whose message starts with a reason token. `create`,
+`read` and `append` read the token and print what to do instead of the token,
+with the server's own detail after it:
+
+- `STREAM_PRODUCER_CONFLICT`: the producer already appended different content
+  at this sequence; nothing was written.
+- `STREAM_PRODUCER_STALE_SEQUENCE`: the sequence is below the producer's
+  latest; nothing was written.
+- `STREAM_CURSOR_BELOW_FLOOR`: the read starts below the floor; those records
+  are gone.
+- `STREAM_CLOSED`: the stream is closed; its records stay readable.
+- `STREAM_POLICY_MISMATCH`: a stream with this id exists with a different
+  lifecycle.
+
+## Payload codec
+
+The records' bodies and metadata go through the same remote codec as every
+other payload the CLI shows or sends, configured with `--codec-endpoint`,
+`--codec-auth` and `--codec-header`. The gRPC interceptor that applies the
+codec walks only the public API's messages, so the stream commands apply it by
+hand on the records and on a close reason.
+
+## Examples
+
+```sh
+temporal server start-dev --port 7533
+
+temporal stream create --address localhost:7533 --stream-id scores
+temporal stream append --address localhost:7533 --stream-id scores \
+    --producer-id game --sequence 1 --input '{"home": 1}' --input '{"home": 2}'
+temporal stream read --address localhost:7533 --stream-id scores --follow
+temporal stream describe --address localhost:7533 --stream-id scores
+
+temporal stream read --address localhost:7533 \
+    --workflow-id june-s1-abcdef12 --name scores
+```

internal/devserver/server.go

@@ -41,6 +41,7 @@ import (
 	"go.temporal.io/api/enums/v1"
 	"go.temporal.io/server/chasm/lib/activity"
 	chasmcallback "go.temporal.io/server/chasm/lib/callback"
+	"go.temporal.io/server/chasm/lib/stream"
 	"go.temporal.io/server/common/authorization"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/config"
@@ -251,6 +252,10 @@ func (s *StartOptions) buildServerOptions() ([]temporal.ServerOption, *slog.Leve
 	dynConf[activity.EnableStandaloneActivityOperatorCommands.Key()] = true
 	dynConf[dynamicconfig.FrontendEnableBatchOperationsForStandaloneActivities.Key()] = true
 
+	// The server ships streams off so a deployment opts in. The dev server is
+	// the opt-in: local development is what it exists for, and an explicit
+	// value below still wins.
+	dynConf[stream.EnabledSetting.Key()] = true
 	// The server calls no callback address until one is allowed. A local
 	// receiver is the usual target of a notification channel's callback on
 	// a dev server, and it rarely serves TLS.

internal/temporalcli/client.go

@@ -7,6 +7,7 @@ import (
 	"io/fs"
 	"os"
 	"os/user"
+	"sync/atomic"
 
 	"github.com/temporalio/cli/cliext"
 	"go.temporal.io/api/common/v1"
@@ -32,8 +33,30 @@ func dialClient(cctx *CommandContext, c *cliext.ClientOptions) (client.Client, e
 // used by the gRPC interceptor; callers can use it to decode payloads nested inside
 // opaque proto bytes (e.g. the request/response of a system Nexus operation).
 func dialClientWithCodec(cctx *CommandContext, c *cliext.ClientOptions) (client.Client, converter.PayloadCodec, error) {
+	cl, err := dialClientWithConn(cctx, c)
+	if err != nil {
+		return nil, nil, err
+	}
+	return cl.Client, cl.PayloadCodec, nil
+}
+
+// dialedClient is an SDK client together with what the SDK dialed for it.
+type dialedClient struct {
+	client.Client
+	// The connection the SDK dialed, carrying its TLS, credentials, headers and
+	// interceptors, for services the SDK client does not wrap.
+	Conn grpc.ClientConnInterface
+	// As returned by [dialClientWithCodec]. The interceptor applying it walks
+	// the public API's messages only, so payloads inside any other message are
+	// encoded and decoded by hand.
+	PayloadCodec converter.PayloadCodec
+}
+
+// dialClientWithConn is like [dialClient] but also returns the connection the
+// SDK dialed and the remote payload codec.
+func dialClientWithConn(cctx *CommandContext, c *cliext.ClientOptions) (*dialedClient, error) {
 	if cctx.RootCommand == nil {
-		return nil, nil, fmt.Errorf("root command unexpectedly missing when dialing client")
+		return nil, fmt.Errorf("root command unexpectedly missing when dialing client")
 	}
 
 	// Set default identity if not provided
@@ -61,12 +84,12 @@ func dialClientWithCodec(cctx *CommandContext, c *cliext.ClientOptions) (client.
 		// original setup error instead of attaching a guessed address or profile.
 		var pathErr *fs.PathError
 		if errors.As(err, &pathErr) {
-			return nil, nil, newConnectError(&connectDiagnosis{
+			return nil, newConnectError(&connectDiagnosis{
 				Cause:  causeCertFileUnreadable,
 				Detail: pathErr.Path,
 			}, connectMeta{}, err)
 		}
-		return nil, nil, err
+		return nil, err
 	}
 
 	// We do not put codec on data converter here, it is applied via
@@ -82,6 +105,20 @@ func dialClientWithCodec(cctx *CommandContext, c *cliext.ClientOptions) (client.
 	clientOpts.ConnectionOptions.DialOptions = append(
 		clientOpts.ConnectionOptions.DialOptions, grpc.WithChainUnaryInterceptor(fixedHeaderOverrideInterceptor))
 
+	// The SDK does not hand out the connection it dials, but every interceptor
+	// on it is given the connection, and the dial itself makes a call.
+	var conn atomic.Pointer[grpc.ClientConn]
+	clientOpts.ConnectionOptions.DialOptions = append(
+		clientOpts.ConnectionOptions.DialOptions, grpc.WithChainUnaryInterceptor(
+			func(
+				ctx context.Context,
+				method string, req, reply any,
+				cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption,
+			) error {
+				conn.CompareAndSwap(nil, cc)
+				return invoker(ctx, method, req, reply, cc, opts...)
+			}))
+
 	// Additional gRPC options
 	clientOpts.ConnectionOptions.DialOptions = append(
 		clientOpts.ConnectionOptions.DialOptions, cctx.Options.AdditionalClientGRPCDialOptions...)
@@ -97,14 +134,19 @@ func dialClientWithCodec(cctx *CommandContext, c *cliext.ClientOptions) (client.
 
 	cl, err := client.DialContext(dialCtx, clientOpts)
 	if err != nil {
-		return nil, nil, dialConnectError(cctx, dialCtx, clientOpts, err)
+		return nil, dialConnectError(cctx, dialCtx, clientOpts, err)
 	}
 
 	// Since this namespace value is used by many commands after this call,
 	// we are mutating it to be the derived one
 	c.Namespace = clientOpts.Namespace
 
-	return cl, builder.PayloadCodec, nil
+	cc := conn.Load()
+	if cc == nil {
+		cl.Close()
+		return nil, fmt.Errorf("dialing made no call, so the connection was not captured")
+	}
+	return &dialedClient{Client: cl, Conn: cc, PayloadCodec: builder.PayloadCodec}, nil
 }
 
 // dialConnectError enriches a client.DialContext failure with a staged

internal/temporalcli/commands.channel_test.go

@@ -24,6 +24,7 @@ import (
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/workflow"
+	streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -144,6 +145,14 @@ func (f *fakeChannelService) PollChannel(
 }
 
 func startFakeChannelService(t *testing.T, f *fakeChannelService) string {
+	return startFakeServices(t, f, nil)
+}
+
+// startFakeServices serves a fake workflow service and, when given, a fake
+// stream service on one address, the way the Service's frontend does.
+func startFakeServices(
+	t *testing.T, wf workflowservice.WorkflowServiceServer, st streampb.StreamServiceServer,
+) string {
 	ln, err := net.Listen("tcp", "127.0.0.1:0")
 	require.NoError(t, err)
 	// The Service sends its errors as gRPC statuses; a bare server would send
@@ -157,7 +166,10 @@ func startFakeChannelService(t *testing.T, f *fakeChannelService) string {
 		}
 		return resp, nil
 	}))
-	workflowservice.RegisterWorkflowServiceServer(srv, f)
+	workflowservice.RegisterWorkflowServiceServer(srv, wf)
+	if st != nil {
+		streampb.RegisterStreamServiceServer(srv, st)
+	}
 	go func() { _ = srv.Serve(ln) }()
 	t.Cleanup(srv.Stop)
 	return ln.Addr().String()
@@ -1072,6 +1084,28 @@ func (s *SharedServerSuite) TestChannel_LinkedToActivity() {
 	res = channel("listener", "remove", "--listener-id", added.GetListenerId())
 	s.NoError(res.Err)
 
+	// The activity's own stream notifies the linked channel named for the
+	// stream alone, as the describe card says it will.
+	res = s.Execute("stream", "append", "--address", s.Address(), "--activity-id", activityID,
+		"--name", "scores", "--input", `{"home": 1}`)
+	s.NoError(res.Err)
+	res = s.Execute("stream", "describe", "--address", s.Address(), "--activity-id", activityID,
+		"--name", "scores")
+	s.NoError(res.Err)
+	s.ContainsOnSameLine(res.Stdout.String(), "Channel", "stream/scores")
+	s.ContainsOnSameLine(res.Stdout.String(), "Kind", "Linked")
+	s.ContainsOnSameLine(res.Stdout.String(), "LinkedTo", "activity "+activityID)
+	res = s.Execute("channel", "poll", "--address", s.Address(), "-c", "stream/scores",
+		"--activity-id", activityID, "--after-counter", "0", "--wait", "10s", "-o", "jsonl")
+	s.NoError(res.Err)
+	raw = decodeJSONValues(s.T(), res.Stdout.String())
+	s.NotEmpty(raw)
+	var streamed notificationpb.Notification
+	s.NoError(temporalcli.UnmarshalProtoJSONWithOptions(raw[len(raw)-1], &streamed, true))
+	s.Equal(runID+":1", string(streamed.GetPosition()))
+	s.Equal(enumspb.EXECUTION_TYPE_ACTIVITY, streamed.GetLinkedTo().GetType())
+	s.Equal(activityID, streamed.GetLinkedTo().GetBusinessId())
+
 	// An activity nobody started has no channels to reach.
 	res = s.Execute("channel", "describe", "--address", s.Address(), "-c", ch,
 		"--activity-id", "never-"+uuid.NewString())