diff --git a/docs/stream.md b/docs/stream.md new file mode 100644 index 000000000..ca5bf1d3a --- /dev/null +++ b/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 ` +: 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 [--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 --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 +``` diff --git a/internal/devserver/server.go b/internal/devserver/server.go index 5ef689a7e..a4876fadf 100644 --- a/internal/devserver/server.go +++ b/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. diff --git a/internal/temporalcli/client.go b/internal/temporalcli/client.go index 9126a2b0f..80e70f2a1 100644 --- a/internal/temporalcli/client.go +++ b/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 diff --git a/internal/temporalcli/commands.channel_test.go b/internal/temporalcli/commands.channel_test.go index 800708649..f598daaf9 100644 --- a/internal/temporalcli/commands.channel_test.go +++ b/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()) diff --git a/internal/temporalcli/commands.gen.go b/internal/temporalcli/commands.gen.go index ed56d7cc1..59cc77623 100644 --- a/internal/temporalcli/commands.gen.go +++ b/internal/temporalcli/commands.gen.go @@ -114,6 +114,35 @@ func (v *ChannelOptions) BuildFlags(f *pflag.FlagSet) { f.StringVarP(&v.RunId, "run-id", "r", "", "Run ID of the linked execution. Defaults to the current run. Requires --workflow-id or --activity-id.") } +type StreamIdOptions struct { + StreamId string + FlagSet *pflag.FlagSet +} + +func (v *StreamIdOptions) BuildFlags(f *pflag.FlagSet) { + v.FlagSet = f + f.StringVarP(&v.StreamId, "stream-id", "s", "", "Stream ID of a standalone Stream. Required.") + _ = cobra.MarkFlagRequired(f, "stream-id") +} + +type StreamReferenceOptions struct { + StreamId string + WorkflowId string + RunId string + ActivityId string + Name string + FlagSet *pflag.FlagSet +} + +func (v *StreamReferenceOptions) BuildFlags(f *pflag.FlagSet) { + v.FlagSet = f + f.StringVarP(&v.StreamId, "stream-id", "s", "", "Stream ID of a standalone Stream. You must set either --stream-id or an owner.") + f.StringVarP(&v.WorkflowId, "workflow-id", "w", "", "Workflow ID of the Workflow Execution owning the Stream. Cannot use with --stream-id.") + f.StringVarP(&v.RunId, "run-id", "r", "", "Run ID of the owner. Defaults to the current run.") + f.StringVar(&v.ActivityId, "activity-id", "", "Activity ID of the Activity owning the Stream. With --workflow-id it names an Activity the Workflow scheduled. On its own it names a standalone Activity.") + f.StringVar(&v.Name, "name", "", "Name of the Stream within its owner. Defaults to the owner's default Stream.") +} + type DeploymentNameOptions struct { Name string FlagSet *pflag.FlagSet @@ -554,6 +583,7 @@ func NewTemporalCommand(cctx *CommandContext) *TemporalCommand { s.Command.AddCommand(&NewTemporalOperatorCommand(cctx, &s).Command) s.Command.AddCommand(&NewTemporalScheduleCommand(cctx, &s).Command) s.Command.AddCommand(&NewTemporalServerCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalStreamCommand(cctx, &s).Command) s.Command.AddCommand(&NewTemporalTaskQueueCommand(cctx, &s).Command) s.Command.AddCommand(&NewTemporalWorkerCommand(cctx, &s).Command) s.Command.AddCommand(&NewTemporalWorkflowCommand(cctx, &s).Command) @@ -3073,9 +3103,9 @@ func NewTemporalServerStartDevCommand(cctx *CommandContext, parent *TemporalServ s.Command.Use = "start-dev [flags]" s.Command.Short = "Start Temporal development server" if hasHighlighting { - s.Command.Long = "Run a development Temporal Server on your local system.\n\n\x1b[1m+------------------------------------------------------------------------+\n| WARNING: The development server is not intended for production use. |\n| It skips certain HTTP security checks to make local use simpler. |\n| |\n| For production use, see: |\n| https://docs.temporal.io/production-deployment |\n+------------------------------------------------------------------------+\x1b[0m\n\nView the Web UI for the default configuration at: http://localhost:8233\n\n\x1b[1mtemporal server start-dev\x1b[0m\n\nAdd persistence for Workflow Executions across runs:\n\n\x1b[1mtemporal server start-dev \\\n --db-filename path-to-your-local-persistent-store\x1b[0m\n\nSet the port from the front-end gRPC Service (7233 default):\n\n\x1b[1mtemporal server start-dev \\\n --port 7000\x1b[0m\n\nUse a custom port for the Web UI. The default is the gRPC port (7233 default)\nplus 1000 (8233):\n\n\x1b[1mtemporal server start-dev \\\n --ui-port 3000\x1b[0m\n\nCallbacks to \x1b[1m127.0.0.1\x1b[0m and \x1b[1mlocalhost\x1b[0m on any port are allowed over\nplain HTTP, so a local receiver can listen on a notification channel.\nReplace the list with a dynamic configuration value:\n\n\x1b[1mtemporal server start-dev \\\n --dynamic-config-value 'callback.allowedAddresses=[]'\x1b[0m" + s.Command.Long = "Run a development Temporal Server on your local system.\n\n\x1b[1m+------------------------------------------------------------------------+\n| WARNING: The development server is not intended for production use. |\n| It skips certain HTTP security checks to make local use simpler. |\n| |\n| For production use, see: |\n| https://docs.temporal.io/production-deployment |\n+------------------------------------------------------------------------+\x1b[0m\n\nView the Web UI for the default configuration at: http://localhost:8233\n\n\x1b[1mtemporal server start-dev\x1b[0m\n\nAdd persistence for Workflow Executions across runs:\n\n\x1b[1mtemporal server start-dev \\\n --db-filename path-to-your-local-persistent-store\x1b[0m\n\nSet the port from the front-end gRPC Service (7233 default):\n\n\x1b[1mtemporal server start-dev \\\n --port 7000\x1b[0m\n\nUse a custom port for the Web UI. The default is the gRPC port (7233 default)\nplus 1000 (8233):\n\n\x1b[1mtemporal server start-dev \\\n --ui-port 3000\x1b[0m\n\nStreams are enabled on the development server. Turn them off with a\ndynamic configuration value:\n\n\x1b[1mtemporal server start-dev \\\n --dynamic-config-value 'stream.enabled=false'\x1b[0m\n\nCallbacks to \x1b[1m127.0.0.1\x1b[0m and \x1b[1mlocalhost\x1b[0m on any port are allowed over\nplain HTTP, so a local receiver can listen on a notification channel.\nReplace the list with a dynamic configuration value:\n\n\x1b[1mtemporal server start-dev \\\n --dynamic-config-value 'callback.allowedAddresses=[]'\x1b[0m" } else { - s.Command.Long = "Run a development Temporal Server on your local system.\n\n```\n+------------------------------------------------------------------------+\n| WARNING: The development server is not intended for production use. |\n| It skips certain HTTP security checks to make local use simpler. |\n| |\n| For production use, see: |\n| https://docs.temporal.io/production-deployment |\n+------------------------------------------------------------------------+\n```\n\nView the Web UI for the default configuration at: http://localhost:8233\n\n```\ntemporal server start-dev\n```\n\nAdd persistence for Workflow Executions across runs:\n\n```\ntemporal server start-dev \\\n --db-filename path-to-your-local-persistent-store\n```\n\nSet the port from the front-end gRPC Service (7233 default):\n\n```\ntemporal server start-dev \\\n --port 7000\n```\n\nUse a custom port for the Web UI. The default is the gRPC port (7233 default)\nplus 1000 (8233):\n\n```\ntemporal server start-dev \\\n --ui-port 3000\n```\n\nCallbacks to `127.0.0.1` and `localhost` on any port are allowed over\nplain HTTP, so a local receiver can listen on a notification channel.\nReplace the list with a dynamic configuration value:\n\n```\ntemporal server start-dev \\\n --dynamic-config-value 'callback.allowedAddresses=[]'\n```" + s.Command.Long = "Run a development Temporal Server on your local system.\n\n```\n+------------------------------------------------------------------------+\n| WARNING: The development server is not intended for production use. |\n| It skips certain HTTP security checks to make local use simpler. |\n| |\n| For production use, see: |\n| https://docs.temporal.io/production-deployment |\n+------------------------------------------------------------------------+\n```\n\nView the Web UI for the default configuration at: http://localhost:8233\n\n```\ntemporal server start-dev\n```\n\nAdd persistence for Workflow Executions across runs:\n\n```\ntemporal server start-dev \\\n --db-filename path-to-your-local-persistent-store\n```\n\nSet the port from the front-end gRPC Service (7233 default):\n\n```\ntemporal server start-dev \\\n --port 7000\n```\n\nUse a custom port for the Web UI. The default is the gRPC port (7233 default)\nplus 1000 (8233):\n\n```\ntemporal server start-dev \\\n --ui-port 3000\n```\n\nStreams are enabled on the development server. Turn them off with a\ndynamic configuration value:\n\n```\ntemporal server start-dev \\\n --dynamic-config-value 'stream.enabled=false'\n```\n\nCallbacks to `127.0.0.1` and `localhost` on any port are allowed over\nplain HTTP, so a local receiver can listen on a notification channel.\nReplace the list with a dynamic configuration value:\n\n```\ntemporal server start-dev \\\n --dynamic-config-value 'callback.allowedAddresses=[]'\n```" } s.Command.Args = cobra.NoArgs s.Command.Flags().StringVarP(&s.DbFilename, "db-filename", "f", "", "Path to file for persistent Temporal state store. By default, Workflow Executions are lost when the server process dies.") @@ -3103,6 +3133,298 @@ func NewTemporalServerStartDevCommand(cctx *CommandContext, parent *TemporalServ return &s } +type TemporalStreamCommand struct { + Parent *TemporalCommand + Command cobra.Command + cliext.ClientOptions +} + +func NewTemporalStreamCommand(cctx *CommandContext, parent *TemporalCommand) *TemporalStreamCommand { + var s TemporalStreamCommand + s.Parent = parent + s.Command.Use = "stream" + s.Command.Short = "Read and manage Streams" + if hasHighlighting { + s.Command.Long = "A Stream is an append-only log the Temporal Service carries next to\nWorkflow Executions. A standalone Stream has an ID of its own. An owned\nStream lives inside a Workflow Execution or an Activity and is addressed\nby its owner and a Stream name:\n\n\x1b[1mtemporal stream [command] [options]\x1b[0m\n\nFor example:\n\n\x1b[1mtemporal stream read \\\n --workflow-id YourWorkflowId\x1b[0m\n\nThe Service must enable Streams for the Namespace through the\n\x1b[1mstream.enabled\x1b[0m dynamic configuration value. The development server\nenables them." + } else { + s.Command.Long = "A Stream is an append-only log the Temporal Service carries next to\nWorkflow Executions. A standalone Stream has an ID of its own. An owned\nStream lives inside a Workflow Execution or an Activity and is addressed\nby its owner and a Stream name:\n\n```\ntemporal stream [command] [options]\n```\n\nFor example:\n\n```\ntemporal stream read \\\n --workflow-id YourWorkflowId\n```\n\nThe Service must enable Streams for the Namespace through the\n`stream.enabled` dynamic configuration value. The development server\nenables them." + } + s.Command.Args = cobra.NoArgs + s.Command.AddCommand(&NewTemporalStreamAppendCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalStreamCloseCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalStreamCreateCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalStreamDeleteCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalStreamDescribeCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalStreamListCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalStreamReadCommand(cctx, &s).Command) + s.Command.AddCommand(&NewTemporalStreamTruncateCommand(cctx, &s).Command) + s.ClientOptions.BuildFlags(s.Command.PersistentFlags()) + s.ClientOptions.HideFlags() + return &s +} + +type TemporalStreamAppendCommand struct { + Parent *TemporalStreamCommand + Command cobra.Command + StreamReferenceOptions + PayloadInputOptions + Topic string + ProducerId string + Attempt int + Sequence int + ExpectedOffset int + Finish bool +} + +func NewTemporalStreamAppendCommand(cctx *CommandContext, parent *TemporalStreamCommand) *TemporalStreamAppendCommand { + var s TemporalStreamAppendCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "append [flags]" + s.Command.Short = "Append records to a Stream" + if hasHighlighting { + s.Command.Long = "Append one or more records to a Stream. Each \x1b[1m--input\x1b[0m is one record.\nWithout \x1b[1m--input\x1b[0m or \x1b[1m--input-file\x1b[0m, one record is read per line from\nstdin:\n\n\x1b[1mtemporal stream append \\\n --stream-id YourStreamId \\\n --input '{\"score\": 1}' \\\n --input '{\"score\": 2}'\x1b[0m\n\nAppend to a Stream a Workflow Execution owns, naming the owner instead:\n\n\x1b[1mtemporal stream append \\\n --workflow-id YourWorkflowId \\\n --name scores \\\n --input '{\"score\": 1}'\x1b[0m\n\nSet \x1b[1m--producer-id\x1b[0m and \x1b[1m--sequence\x1b[0m so a retried append is deduplicated\nrather than appended twice. Pass \x1b[1m--finish\x1b[0m to end the producer's writes\non the topic with a \x1b[1mFINISH\x1b[0m record after the values." + } else { + s.Command.Long = "Append one or more records to a Stream. Each `--input` is one record.\nWithout `--input` or `--input-file`, one record is read per line from\nstdin:\n\n```\ntemporal stream append \\\n --stream-id YourStreamId \\\n --input '{\"score\": 1}' \\\n --input '{\"score\": 2}'\n```\n\nAppend to a Stream a Workflow Execution owns, naming the owner instead:\n\n```\ntemporal stream append \\\n --workflow-id YourWorkflowId \\\n --name scores \\\n --input '{\"score\": 1}'\n```\n\nSet `--producer-id` and `--sequence` so a retried append is deduplicated\nrather than appended twice. Pass `--finish` to end the producer's writes\non the topic with a `FINISH` record after the values." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().StringVar(&s.Topic, "topic", "", "Topic label stored on every record.") + s.Command.Flags().StringVar(&s.ProducerId, "producer-id", "", "Producer identity stored on every record. Together with --sequence it deduplicates a retried append.") + s.Command.Flags().IntVar(&s.Attempt, "attempt", 0, "Producer attempt stored on every record.") + s.Command.Flags().IntVar(&s.Sequence, "sequence", 0, "Sequence of the first record within the producer's attempt. Records that follow take the next numbers.") + s.Command.Flags().IntVar(&s.ExpectedOffset, "expected-offset", 0, "Refuse the append unless the Stream's next offset equals this value. Only applies to a standalone Stream.") + s.Command.Flags().BoolVar(&s.Finish, "finish", false, "Append a `FINISH` record after the values.") + s.StreamReferenceOptions.BuildFlags(s.Command.Flags()) + s.PayloadInputOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalStreamCloseCommand struct { + Parent *TemporalStreamCommand + Command cobra.Command + StreamIdOptions + Reason string +} + +func NewTemporalStreamCloseCommand(cctx *CommandContext, parent *TemporalStreamCommand) *TemporalStreamCloseCommand { + var s TemporalStreamCloseCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "close [flags]" + s.Command.Short = "Close a Stream" + if hasHighlighting { + s.Command.Long = "Close a standalone Stream. A closed Stream accepts no more records and\nstays readable until its retention passes:\n\n\x1b[1mtemporal stream close \\\n --stream-id YourStreamId \\\n --reason \"done\"\x1b[0m" + } else { + s.Command.Long = "Close a standalone Stream. A closed Stream accepts no more records and\nstays readable until its retention passes:\n\n```\ntemporal stream close \\\n --stream-id YourStreamId \\\n --reason \"done\"\n```" + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().StringVar(&s.Reason, "reason", "", "Reason recorded on the Stream.") + s.StreamIdOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalStreamCreateCommand struct { + Parent *TemporalStreamCommand + Command cobra.Command + StreamIdOptions + Retention cliext.FlagDuration + MaxItems int + MaxBytes int +} + +func NewTemporalStreamCreateCommand(cctx *CommandContext, parent *TemporalStreamCommand) *TemporalStreamCreateCommand { + var s TemporalStreamCreateCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "create [flags]" + s.Command.Short = "Create a standalone Stream" + if hasHighlighting { + s.Command.Long = "Create a standalone Stream with an ID of its own:\n\n\x1b[1mtemporal stream create \\\n --stream-id YourStreamId\x1b[0m\n\nRecords older than \x1b[1m--retention\x1b[0m are reclaimed, and a closed Stream is\ndeleted once it passes. Without it the Namespace's retention applies.\nCreating a Stream that already exists with the same lifecycle is an\nidempotent retry; a different lifecycle is refused." + } else { + s.Command.Long = "Create a standalone Stream with an ID of its own:\n\n```\ntemporal stream create \\\n --stream-id YourStreamId\n```\n\nRecords older than `--retention` are reclaimed, and a closed Stream is\ndeleted once it passes. Without it the Namespace's retention applies.\nCreating a Stream that already exists with the same lifecycle is an\nidempotent retry; a different lifecycle is refused." + } + s.Command.Args = cobra.NoArgs + s.Retention = 0 + s.Command.Flags().Var(&s.Retention, "retention", "How long a record stays readable, and how long a closed Stream stays readable before it is deleted. Cannot be longer than the Namespace's retention.") + s.Command.Flags().IntVar(&s.MaxItems, "max-items", 0, "Most records the Stream holds before the oldest are reclaimed. Default is zero (unbounded).") + s.Command.Flags().IntVar(&s.MaxBytes, "max-bytes", 0, "Most bytes the Stream holds. An append past it is refused until records are reclaimed. Default is zero (unbounded).") + s.StreamIdOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalStreamDeleteCommand struct { + Parent *TemporalStreamCommand + Command cobra.Command + StreamIdOptions + Force bool +} + +func NewTemporalStreamDeleteCommand(cctx *CommandContext, parent *TemporalStreamCommand) *TemporalStreamDeleteCommand { + var s TemporalStreamDeleteCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "delete [flags]" + s.Command.Short = "Remove a Stream" + if hasHighlighting { + s.Command.Long = "Delete a standalone Stream and every record it holds:\n\n\x1b[1mtemporal stream delete \\\n --stream-id YourStreamId\x1b[0m\n\nThe call is refused while a Workflow Execution consumes the Stream,\nbecause its replay depends on the records. Pass \x1b[1m--force\x1b[0m to delete it\nanyway." + } else { + s.Command.Long = "Delete a standalone Stream and every record it holds:\n\n```\ntemporal stream delete \\\n --stream-id YourStreamId\n```\n\nThe call is refused while a Workflow Execution consumes the Stream,\nbecause its replay depends on the records. Pass `--force` to delete it\nanyway." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().BoolVar(&s.Force, "force", false, "Delete even while a Workflow Execution consumes the Stream.") + s.StreamIdOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalStreamDescribeCommand struct { + Parent *TemporalStreamCommand + Command cobra.Command + StreamReferenceOptions +} + +func NewTemporalStreamDescribeCommand(cctx *CommandContext, parent *TemporalStreamCommand) *TemporalStreamDescribeCommand { + var s TemporalStreamDescribeCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "describe [flags]" + s.Command.Short = "Show a Stream's state" + if hasHighlighting { + s.Command.Long = "Show a Stream's frontier, floor, close state, retention, producers and\nconsumers:\n\n\x1b[1mtemporal stream describe \\\n --stream-id YourStreamId\x1b[0m\n\nDescribe a Stream a Workflow Execution owns:\n\n\x1b[1mtemporal stream describe \\\n --workflow-id YourWorkflowId \\\n --name scores\x1b[0m\n\nThe output also names the notification channel the Stream notifies on\neach append and on its close. The name follows from the Stream alone,\nso no call to the Service is needed. A Stream a Workflow Execution\nowns notifies \x1b[1mstream/NAME\x1b[0m, linked to the owner. A Stream of an\nActivity the Workflow scheduled notifies \x1b[1mstream/ACTIVITY_ID/NAME\x1b[0m,\nlinked to the owner too. A standalone Stream notifies the independent\nchannel \x1b[1mstream/STREAM_ID\x1b[0m. Pass the channel, and the owner of a\nlinked one, to \x1b[1mtemporal channel poll\x1b[0m to follow the Stream:\n\n\x1b[1mtemporal channel poll \\\n --channel stream/scores \\\n --workflow-id YourWorkflowId\x1b[0m" + } else { + s.Command.Long = "Show a Stream's frontier, floor, close state, retention, producers and\nconsumers:\n\n```\ntemporal stream describe \\\n --stream-id YourStreamId\n```\n\nDescribe a Stream a Workflow Execution owns:\n\n```\ntemporal stream describe \\\n --workflow-id YourWorkflowId \\\n --name scores\n```\n\nThe output also names the notification channel the Stream notifies on\neach append and on its close. The name follows from the Stream alone,\nso no call to the Service is needed. A Stream a Workflow Execution\nowns notifies `stream/NAME`, linked to the owner. A Stream of an\nActivity the Workflow scheduled notifies `stream/ACTIVITY_ID/NAME`,\nlinked to the owner too. A standalone Stream notifies the independent\nchannel `stream/STREAM_ID`. Pass the channel, and the owner of a\nlinked one, to `temporal channel poll` to follow the Stream:\n\n```\ntemporal channel poll \\\n --channel stream/scores \\\n --workflow-id YourWorkflowId\n```" + } + s.Command.Args = cobra.NoArgs + s.StreamReferenceOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalStreamListCommand struct { + Parent *TemporalStreamCommand + Command cobra.Command + Query string + Limit int + PageSize int +} + +func NewTemporalStreamListCommand(cctx *CommandContext, parent *TemporalStreamCommand) *TemporalStreamListCommand { + var s TemporalStreamListCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "list [flags]" + s.Command.Short = "Show standalone Streams" + if hasHighlighting { + s.Command.Long = "List the standalone Streams in a Namespace. The optional \x1b[1m--query\x1b[0m\nlimits the output to Streams matching a List Filter, where \x1b[1mWorkflowId\x1b[0m\nholds the Stream ID:\n\n\x1b[1mtemporal stream list \\\n --query 'WorkflowId STARTS_WITH \"scores-\"'\x1b[0m\n\nStreams a Workflow Execution owns are not listed here. Describe or read\nthem through their owner." + } else { + s.Command.Long = "List the standalone Streams in a Namespace. The optional `--query`\nlimits the output to Streams matching a List Filter, where `WorkflowId`\nholds the Stream ID:\n\n```\ntemporal stream list \\\n --query 'WorkflowId STARTS_WITH \"scores-\"'\n```\n\nStreams a Workflow Execution owns are not listed here. Describe or read\nthem through their owner." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().StringVarP(&s.Query, "query", "q", "", "Content for an SQL-like `QUERY` List Filter.") + s.Command.Flags().IntVar(&s.Limit, "limit", 0, "Maximum number of Streams to display.") + s.Command.Flags().IntVar(&s.PageSize, "page-size", 100, "Maximum number of Streams to fetch at a time from the server.") + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalStreamReadCommand struct { + Parent *TemporalStreamCommand + Command cobra.Command + StreamReferenceOptions + FromOffset int + FromTail bool + Last int + Follow bool + Topic []string + Limit int + PageSize int +} + +func NewTemporalStreamReadCommand(cctx *CommandContext, parent *TemporalStreamCommand) *TemporalStreamReadCommand { + var s TemporalStreamReadCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "read [flags]" + s.Command.Short = "Read a Stream's records" + if hasHighlighting { + s.Command.Long = "Read a Stream from the beginning and stop once caught up:\n\n\x1b[1mtemporal stream read \\\n --stream-id YourStreamId\x1b[0m\n\nRead a Stream a Workflow Execution owns and keep following it as records\nland:\n\n\x1b[1mtemporal stream read \\\n --workflow-id YourWorkflowId \\\n --name scores \\\n --follow\x1b[0m\n\nStart at a cursor with \x1b[1m--from-offset\x1b[0m, at the tail with \x1b[1m--from-tail\x1b[0m,\nor at the newest records with \x1b[1m--last\x1b[0m. Each record prints with its\noffset, kind, topic, producer, attempt, sequence and body. With\n\x1b[1m--output json\x1b[0m each record is one JSON object." + } else { + s.Command.Long = "Read a Stream from the beginning and stop once caught up:\n\n```\ntemporal stream read \\\n --stream-id YourStreamId\n```\n\nRead a Stream a Workflow Execution owns and keep following it as records\nland:\n\n```\ntemporal stream read \\\n --workflow-id YourWorkflowId \\\n --name scores \\\n --follow\n```\n\nStart at a cursor with `--from-offset`, at the tail with `--from-tail`,\nor at the newest records with `--last`. Each record prints with its\noffset, kind, topic, producer, attempt, sequence and body. With\n`--output json` each record is one JSON object." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().IntVar(&s.FromOffset, "from-offset", 0, "Offset to start at, inclusive. Refused when below the Stream's floor.") + s.Command.Flags().BoolVar(&s.FromTail, "from-tail", false, "Start at the tail and show only records appended from now on.") + s.Command.Flags().IntVar(&s.Last, "last", 0, "Start at the newest records, showing at most this many first.") + s.Command.Flags().BoolVarP(&s.Follow, "follow", "f", false, "Keep reading as records land, until the Stream closes.") + s.Command.Flags().StringArrayVar(&s.Topic, "topic", nil, "Show only records on this topic. Can be passed multiple times.") + s.Command.Flags().IntVar(&s.Limit, "limit", 0, "Maximum number of records to display.") + s.Command.Flags().IntVar(&s.PageSize, "page-size", 100, "Maximum number of records to fetch at a time from the server.") + s.StreamReferenceOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + +type TemporalStreamTruncateCommand struct { + Parent *TemporalStreamCommand + Command cobra.Command + StreamIdOptions + To int +} + +func NewTemporalStreamTruncateCommand(cctx *CommandContext, parent *TemporalStreamCommand) *TemporalStreamTruncateCommand { + var s TemporalStreamTruncateCommand + s.Parent = parent + s.Command.DisableFlagsInUseLine = true + s.Command.Use = "truncate [flags]" + s.Command.Short = "Drop the oldest records of a Stream" + if hasHighlighting { + s.Command.Long = "Move the readable floor of a standalone Stream up to an offset. Records\nbelow it are gone:\n\n\x1b[1mtemporal stream truncate \\\n --stream-id YourStreamId \\\n --to 100\x1b[0m\n\nThe call is refused while a Workflow Execution consumes records below\nthe new floor." + } else { + s.Command.Long = "Move the readable floor of a standalone Stream up to an offset. Records\nbelow it are gone:\n\n```\ntemporal stream truncate \\\n --stream-id YourStreamId \\\n --to 100\n```\n\nThe call is refused while a Workflow Execution consumes records below\nthe new floor." + } + s.Command.Args = cobra.NoArgs + s.Command.Flags().IntVar(&s.To, "to", 0, "New floor. Records below this offset are dropped. Required.") + _ = cobra.MarkFlagRequired(s.Command.Flags(), "to") + s.StreamIdOptions.BuildFlags(s.Command.Flags()) + s.Command.Run = func(c *cobra.Command, args []string) { + if err := s.run(cctx, args); err != nil { + cctx.Options.Fail(err) + } + } + return &s +} + type TemporalTaskQueueCommand struct { Parent *TemporalCommand Command cobra.Command diff --git a/internal/temporalcli/commands.stream.go b/internal/temporalcli/commands.stream.go new file mode 100644 index 000000000..a6457f096 --- /dev/null +++ b/internal/temporalcli/commands.stream.go @@ -0,0 +1,1020 @@ +package temporalcli + +import ( + "bufio" + "context" + "errors" + "fmt" + "os" + "reflect" + "sort" + "strings" + "time" + + "github.com/fatih/color" + "github.com/mattn/go-isatty" + "github.com/temporalio/cli/cliext" + "github.com/temporalio/cli/internal/printer" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + notificationpb "go.temporal.io/api/notification/v1" + "go.temporal.io/api/serviceerror" + streamapi "go.temporal.io/api/stream/v1" + "go.temporal.io/sdk/converter" + streamlib "go.temporal.io/server/chasm/lib/stream" + streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" + "google.golang.org/protobuf/types/known/durationpb" +) + +// plainRefusal maps a refusal's reason token to plain language. Every other +// error comes back as it was. +func plainRefusal(err error) error { + var refused *serviceerror.FailedPrecondition + if !errors.As(err, &refused) { + return err + } + reason := streamlib.ReasonOf(refused.Message) + if reason == "" { + return err + } + _, detail, _ := strings.Cut(refused.Message, ": ") + var plain string + switch reason { + case streamlib.ReasonProducerConflict: + plain = "the producer already appended different content at this sequence, so " + + "nothing was written. Use the next sequence, or check that two producers do not " + + "share the id." + case streamlib.ReasonProducerStaleSequence: + plain = "the sequence is below the producer's latest, so nothing was written. A " + + "producer carries one append at a time; retry with its latest sequence or move on " + + "to the next." + case streamlib.ReasonCursorBelowFloor: + plain = "the read starts below the stream's floor, and those records are gone. Start " + + "from the beginning, or at the offset the stream now starts at." + case streamlib.ReasonStreamClosed: + plain = "the stream is closed. Its records stay readable, but nothing more can be " + + "appended." + case streamlib.ReasonPolicyMismatch: + plain = "a stream with this id already exists with a different lifecycle. Use a new " + + "id, or repeat the existing retention, max items and max bytes." + } + return &refusalError{plain: plain, detail: detail, cause: err} +} + +// streamTarget is the stream a command names: a standalone stream by its id, +// or a stream an execution owns by that owner and a name. +type streamTarget struct { + streamID string + owner *streampb.StreamOwner + name string +} + +// resolvedName is the name the server resolves an owned stream to when the +// command leaves it out. +func (t streamTarget) resolvedName() string { + if t.name == "" { + return streamlib.DefaultStreamName + } + return t.name +} + +func (t streamTarget) String() string { + if t.streamID != "" { + return fmt.Sprintf("stream %q", t.streamID) + } + name := t.resolvedName() + switch t.owner.GetKind() { + case streampb.STREAM_OWNER_KIND_WORKFLOW_ACTIVITY: + return fmt.Sprintf("stream %q of activity %q in workflow %q", + name, t.owner.GetActivityId(), t.owner.GetId()) + case streampb.STREAM_OWNER_KIND_ACTIVITY: + return fmt.Sprintf("stream %q of activity %q", name, t.owner.GetId()) + default: + return fmt.Sprintf("stream %q of workflow %q", name, t.owner.GetId()) + } +} + +// streamChannelPrefix starts the name of every notification channel a stream +// notifies. +const streamChannelPrefix = "stream/" + +// streamChannel is the notification channel a stream notifies on each append +// and on its close. The server derives the name from the stream's identity +// alone, so the CLI derives the same name without a call and a listener can +// follow a stream before it exists. +type streamChannel struct { + name string + kind notificationpb.ChannelKind + owner *commonpb.Execution +} + +func (t streamTarget) channel() streamChannel { + if t.streamID != "" { + return streamChannel{ + name: streamChannelPrefix + t.streamID, + kind: notificationpb.CHANNEL_KIND_INDEPENDENT, + } + } + owner := &commonpb.Execution{ + Type: enumspb.EXECUTION_TYPE_WORKFLOW, + BusinessId: t.owner.GetId(), + RunId: t.owner.GetRunId(), + } + switch t.owner.GetKind() { + case streampb.STREAM_OWNER_KIND_WORKFLOW_ACTIVITY: + // The activity's stream notifies the workflow that scheduled it, so the + // activity ID is in the name and the workflow is the owner. + return streamChannel{ + name: streamChannelPrefix + t.owner.GetActivityId() + "/" + t.resolvedName(), + kind: notificationpb.CHANNEL_KIND_LINKED, + owner: owner, + } + case streampb.STREAM_OWNER_KIND_ACTIVITY: + // A standalone activity is an execution that holds linked channels of + // its own, since the server resolves the channel's owner from the + // execution a request names. Its stream's channel then needs no + // activity ID in the name. + owner.Type = enumspb.EXECUTION_TYPE_ACTIVITY + return streamChannel{ + name: streamChannelPrefix + t.resolvedName(), + kind: notificationpb.CHANNEL_KIND_LINKED, + owner: owner, + } + default: + return streamChannel{ + name: streamChannelPrefix + t.resolvedName(), + kind: notificationpb.CHANNEL_KIND_LINKED, + owner: owner, + } + } +} + +func (o *StreamReferenceOptions) target() (streamTarget, error) { + hasOwner := o.WorkflowId != "" || o.ActivityId != "" + switch { + case o.StreamId == "" && !hasOwner: + return streamTarget{}, fmt.Errorf( + "must set either --stream-id or an owner with --workflow-id or --activity-id") + case o.StreamId != "" && hasOwner: + return streamTarget{}, fmt.Errorf("cannot set --stream-id with --workflow-id or --activity-id") + case o.StreamId != "" && (o.RunId != "" || o.Name != ""): + return streamTarget{}, fmt.Errorf("--run-id and --name name an owner's stream, not --stream-id") + case o.StreamId != "": + return streamTarget{streamID: o.StreamId}, nil + } + owner := &streampb.StreamOwner{RunId: o.RunId} + switch { + case o.WorkflowId != "" && o.ActivityId != "": + owner.Kind = streampb.STREAM_OWNER_KIND_WORKFLOW_ACTIVITY + owner.Id, owner.ActivityId = o.WorkflowId, o.ActivityId + case o.WorkflowId != "": + owner.Kind = streampb.STREAM_OWNER_KIND_WORKFLOW + owner.Id = o.WorkflowId + default: + owner.Kind = streampb.STREAM_OWNER_KIND_ACTIVITY + owner.Id = o.ActivityId + } + return streamTarget{owner: owner, name: o.Name}, nil +} + +// streamClient reaches the stream service over the connection the SDK dialed, +// so TLS, credentials and headers are the ones every other command uses. +type streamClient struct { + *dialedClient + svc streampb.StreamServiceClient + namespace string +} + +func dialStreamClient(cctx *CommandContext, c *cliext.ClientOptions) (*streamClient, error) { + cl, err := dialClientWithConn(cctx, c) + if err != nil { + return nil, err + } + return &streamClient{ + dialedClient: cl, + svc: streampb.NewStreamServiceClient(cl.Conn), + namespace: c.Namespace, + }, nil +} + +// applyCodec runs every payload the records carry through one codec call, so a +// page costs one round trip to the codec server rather than one per record. +func (s *streamClient) applyCodec(records []*streampb.StreamRecord, encode bool) error { + if s.PayloadCodec == nil { + return nil + } + fn := s.PayloadCodec.Decode + if encode { + fn = s.PayloadCodec.Encode + } + var payloads []*commonpb.Payload + var put []func(*commonpb.Payload) + for _, r := range records { + if r.Body != nil { + payloads = append(payloads, r.Body) + put = append(put, func(p *commonpb.Payload) { r.Body = p }) + } + for k, v := range r.Metadata { + if v == nil { + continue + } + payloads = append(payloads, v) + put = append(put, func(p *commonpb.Payload) { r.Metadata[k] = p }) + } + } + if len(payloads) == 0 { + return nil + } + out, err := fn(payloads) + if err != nil { + return err + } + if len(out) != len(payloads) { + return fmt.Errorf("codec returned %d payloads for %d", len(out), len(payloads)) + } + for i, p := range out { + put[i](p) + } + return nil +} + +func (s *streamClient) decodeRecords(records []*streampb.StreamRecord) error { + return s.applyCodec(records, false) +} + +func (s *streamClient) encodeRecords(records []*streampb.StreamRecord) error { + return s.applyCodec(records, true) +} + +func (s *streamClient) describe( + ctx context.Context, t streamTarget, +) (*streampb.StreamState, error) { + if t.streamID != "" { + resp, err := s.svc.DescribeStream(ctx, &streampb.DescribeStreamRequest{ + FrontendRequest: &streampb.DescribeStreamInput{ + Namespace: s.namespace, + StreamId: t.streamID, + }, + }) + return resp.GetFrontendResponse().GetState(), err + } + resp, err := s.svc.DescribeWorkflowStream(ctx, &streampb.DescribeWorkflowStreamRequest{ + FrontendRequest: &streampb.DescribeWorkflowStreamInput{ + Namespace: s.namespace, + Owner: t.owner, + StreamName: t.name, + }, + }) + return resp.GetFrontendResponse().GetState(), err +} + +func (s *streamClient) poll( + ctx context.Context, + t streamTarget, + from int64, + start *streamapi.StreamStartPosition, + pageSize int32, + topics []string, + wait bool, +) (*streampb.PollMessagesOutput, error) { + if t.streamID != "" { + resp, err := s.svc.PollMessages(ctx, &streampb.PollMessagesRequest{ + FrontendRequest: &streampb.PollMessagesInput{ + Namespace: s.namespace, + StreamId: t.streamID, + FromOffset: from, + StartPosition: start, + MaxMessages: pageSize, + Topics: topics, + WaitNewMessages: wait, + }, + }) + return resp.GetFrontendResponse(), err + } + resp, err := s.svc.PollWorkflowMessages(ctx, &streampb.PollWorkflowMessagesRequest{ + FrontendRequest: &streampb.PollWorkflowMessagesInput{ + Namespace: s.namespace, + Owner: t.owner, + StreamName: t.name, + FromOffset: from, + StartPosition: start, + MaxMessages: pageSize, + Topics: topics, + WaitNewMessages: wait, + }, + }) + return resp.GetFrontendResponse(), err +} + +func (s *streamClient) append( + ctx context.Context, + t streamTarget, + records []*streampb.StreamRecord, + producerID string, + sequence int64, + expectedOffset *int64, +) (*streampb.AddMessagesOutput, error) { + if t.streamID != "" { + in := &streampb.AddMessagesInput{ + Namespace: s.namespace, + StreamId: t.streamID, + Records: records, + ProducerId: producerID, + Sequence: sequence, + } + if expectedOffset != nil { + in.ExpectedOffset, in.UseExpectedOffset = *expectedOffset, true + } + resp, err := s.svc.AddMessages(ctx, &streampb.AddMessagesRequest{FrontendRequest: in}) + return resp.GetFrontendResponse(), err + } + resp, err := s.svc.AddWorkflowMessages(ctx, &streampb.AddWorkflowMessagesRequest{ + FrontendRequest: &streampb.AddWorkflowMessagesInput{ + Namespace: s.namespace, + Owner: t.owner, + StreamName: t.name, + Records: records, + ProducerId: producerID, + Sequence: sequence, + }, + }) + return resp.GetFrontendResponse(), err +} + +func (c *TemporalStreamCreateCommand) run(cctx *CommandContext, _ []string) error { + if c.MaxItems < 0 || c.MaxBytes < 0 { + return fmt.Errorf("--max-items and --max-bytes cannot be negative") + } + cl, err := dialStreamClient(cctx, &c.Parent.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + lifecycle := &streampb.StreamLifecycle{ + MaxItems: int64(c.MaxItems), + MaxBytes: int64(c.MaxBytes), + } + if c.Retention.Duration() > 0 { + lifecycle.Retention = durationpb.New(c.Retention.Duration()) + } + resp, err := cl.svc.CreateStream(cctx, &streampb.CreateStreamRequest{ + FrontendRequest: &streampb.CreateStreamInput{ + Namespace: cl.namespace, + StreamId: c.StreamId, + Lifecycle: lifecycle, + }, + }) + if err != nil { + return fmt.Errorf("failed creating stream %q: %w", c.StreamId, plainRefusal(err)) + } + if cctx.JSONOutput { + return cctx.Printer.PrintStructured(resp.GetFrontendResponse(), printer.StructuredOptions{}) + } + cctx.Printer.Printlnf("Created stream %v with run id %v", + c.StreamId, resp.GetFrontendResponse().GetRunId()) + return nil +} + +func (c *TemporalStreamListCommand) run(cctx *CommandContext, _ []string) error { + cl, err := dialStreamClient(cctx, &c.Parent.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + // This is a listing command subject to json vs jsonl rules + cctx.Printer.StartList() + defer cctx.Printer.EndList() + + pageSize := c.PageSize + if c.Limit > 0 && c.Limit < pageSize { + pageSize = c.Limit + } + type row struct { + StreamId string + RunId string + } + var token []byte + var seen int + for pageIndex := 0; ; pageIndex++ { + resp, err := cl.svc.ListStreams(cctx, &streampb.ListStreamsRequest{ + FrontendRequest: &streampb.ListStreamsInput{ + Namespace: cl.namespace, + PageSize: int32(pageSize), + NextPageToken: token, + Query: c.Query, + }, + }) + if err != nil { + return fmt.Errorf("failed listing streams: %w", err) + } + page := resp.GetFrontendResponse() + var rows []row + for _, entry := range page.GetStreams() { + if c.Limit > 0 && seen >= c.Limit { + break + } + seen++ + if cctx.JSONOutput { + _ = cctx.Printer.PrintStructured(entry, printer.StructuredOptions{}) + } else { + rows = append(rows, row{StreamId: entry.GetStreamId(), RunId: entry.GetRunId()}) + } + } + // Print table, headers only on first table + if len(rows) > 0 { + _ = cctx.Printer.PrintStructured(rows, printer.StructuredOptions{ + Table: &printer.TableOptions{NoHeader: pageIndex > 0}, + }) + } + token = page.GetNextPageToken() + if len(token) == 0 || (c.Limit > 0 && seen >= c.Limit) { + return nil + } + } +} + +func (c *TemporalStreamDescribeCommand) run(cctx *CommandContext, _ []string) error { + target, err := c.target() + if err != nil { + return err + } + cl, err := dialStreamClient(cctx, &c.Parent.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + state, err := cl.describe(cctx, target) + if err != nil { + return fmt.Errorf("failed describing %v: %w", target, err) + } + if reason := state.GetCloseReason(); reason != nil && cl.PayloadCodec != nil { + decoded, err := cl.PayloadCodec.Decode([]*commonpb.Payload{reason}) + if err != nil { + return fmt.Errorf("failed decoding close reason: %w", err) + } + if len(decoded) == 1 { + state.CloseReason = decoded[0] + } + } + if cctx.JSONOutput { + return cctx.Printer.PrintStructured(state, printer.StructuredOptions{}) + } + + // The frontier, the floor and the close state are facts at zero too, so + // they stay on the card; the rest is shown when set. + info := struct { + StreamId string `cli:",cardOmitEmpty"` + WorkflowId string `cli:",cardOmitEmpty"` + ActivityId string `cli:",cardOmitEmpty"` + RunId string `cli:",cardOmitEmpty"` + Name string `cli:",cardOmitEmpty"` + HeadOffset int64 + BaseOffset int64 + Records int64 + HeldBytes int64 + AppendedBytes int64 + Closed bool + CloseTime time.Time `cli:",cardOmitEmpty"` + CloseReason *commonpb.Payload `cli:",cardOmitEmpty"` + RedirectRunId string `cli:",cardOmitEmpty"` + Retention time.Duration `cli:",cardOmitEmpty"` + MaxItems int64 `cli:",cardOmitEmpty"` + MaxBytes int64 `cli:",cardOmitEmpty"` + BudgetMaxItems int64 `cli:",cardOmitEmpty"` + BudgetMaxBytes int64 `cli:",cardOmitEmpty"` + }{ + StreamId: target.streamID, + HeadOffset: state.GetHeadOffset(), + BaseOffset: state.GetBaseOffset(), + Records: state.GetHeadOffset() - state.GetBaseOffset(), + HeldBytes: state.GetHeldBytes(), + AppendedBytes: state.GetAppendedBytes(), + Closed: state.GetClosed(), + CloseReason: state.GetCloseReason(), + RedirectRunId: state.GetRedirectRunId(), + Retention: state.GetLifecycle().GetRetention().AsDuration(), + MaxItems: state.GetLifecycle().GetMaxItems(), + MaxBytes: state.GetLifecycle().GetMaxBytes(), + BudgetMaxItems: state.GetBudget().GetMaxItems(), + BudgetMaxBytes: state.GetBudget().GetMaxBytes(), + } + if state.GetCloseTime() != nil { + info.CloseTime = state.GetCloseTime().AsTime() + } + if owner := target.owner; owner != nil { + info.RunId, info.Name = owner.GetRunId(), target.name + if owner.GetKind() == streampb.STREAM_OWNER_KIND_ACTIVITY { + info.ActivityId = owner.GetId() + } else { + info.WorkflowId, info.ActivityId = owner.GetId(), owner.GetActivityId() + } + } + cctx.Printer.Println(color.MagentaString("Stream:")) + if err := cctx.Printer.PrintStructured(info, printer.StructuredOptions{}); err != nil { + return err + } + + ch := target.channel() + cctx.Printer.Println() + cctx.Printer.Println(color.MagentaString("Notification Channel:")) + if err := cctx.Printer.PrintStructured(struct { + Channel string + Kind string + LinkedTo string `cli:",cardOmitEmpty"` + }{ + Channel: ch.name, + Kind: ch.kind.String(), + LinkedTo: executionText(ch.owner), + }, printer.StructuredOptions{}); err != nil { + return err + } + + type producerRow struct { + ProducerId string + Sequence int64 + FirstOffset int64 + Count int64 + Fenced bool + } + producers := make([]producerRow, 0, len(state.GetProducers())) + for id, p := range state.GetProducers() { + producers = append(producers, producerRow{ + ProducerId: id, + Sequence: p.GetSeq(), + FirstOffset: p.GetFirstOffset(), + Count: p.GetCount(), + Fenced: p.GetFenced(), + }) + } + sort.Slice(producers, func(i, j int) bool { + return producers[i].ProducerId < producers[j].ProducerId + }) + cctx.Printer.Println() + cctx.Printer.Println(color.MagentaString("Producers: %v", len(producers))) + if len(producers) > 0 { + if err := cctx.Printer.PrintStructured(producers, printer.StructuredOptions{ + Table: &printer.TableOptions{}, + }); err != nil { + return err + } + } + + type consumerRow struct { + ConsumerId string + WorkflowId string + RunId string + Offset int64 + ReplayFloor int64 + Active bool + External bool + } + consumers := make([]consumerRow, 0, len(state.GetConsumers())) + for id, c := range state.GetConsumers() { + consumers = append(consumers, consumerRow{ + ConsumerId: id, + WorkflowId: c.GetWorkflowId(), + RunId: c.GetRunId(), + Offset: c.GetOffset(), + ReplayFloor: c.GetReplayFloor(), + Active: c.GetActive(), + External: c.GetExternal(), + }) + } + sort.Slice(consumers, func(i, j int) bool { + return consumers[i].ConsumerId < consumers[j].ConsumerId + }) + cctx.Printer.Println() + cctx.Printer.Println(color.MagentaString("Consumers: %v", len(consumers))) + if len(consumers) > 0 { + return cctx.Printer.PrintStructured(consumers, printer.StructuredOptions{ + Table: &printer.TableOptions{}, + }) + } + return nil +} + +// startPosition settles where the first poll begins. The beginning is the +// default; an explicit offset of zero is allowed and differs from it once the +// floor has moved. +func (c *TemporalStreamReadCommand) startPosition() (*streamapi.StreamStartPosition, error) { + fromOffset := c.Command.Flags().Changed("from-offset") + set := 0 + for _, on := range []bool{fromOffset, c.FromTail, c.Last != 0} { + if on { + set++ + } + } + if set > 1 { + return nil, fmt.Errorf("set at most one of --from-offset, --from-tail and --last") + } + switch { + case fromOffset: + if c.FromOffset < 0 { + return nil, fmt.Errorf("--from-offset cannot be negative") + } + return &streamapi.StreamStartPosition{ + Position: &streamapi.StreamStartPosition_Offset{Offset: int64(c.FromOffset)}, + }, nil + case c.FromTail: + return &streamapi.StreamStartPosition{ + Position: &streamapi.StreamStartPosition_Tail{Tail: true}, + }, nil + case c.Last != 0: + if c.Last < 0 { + return nil, fmt.Errorf("--last must be positive") + } + return &streamapi.StreamStartPosition{ + Position: &streamapi.StreamStartPosition_LastN{LastN: int64(c.Last)}, + }, nil + default: + return &streamapi.StreamStartPosition{ + Position: &streamapi.StreamStartPosition_Earliest{Earliest: true}, + }, nil + } +} + +// streamReader pages through a stream. Without follow it stops once caught up +// with the frontier; with follow it long-polls until the stream closes. +type streamReader struct { + ctx context.Context + cl *streamClient + target streamTarget + start *streamapi.StreamStartPosition + from int64 + topics []string + follow bool + pageSize int32 + limit int + seen int + buf []*streampb.StreamRecord + done bool +} + +func (r *streamReader) next() (*streampb.StreamRecord, error) { + if r.limit > 0 && r.seen >= r.limit { + return nil, nil + } + for len(r.buf) == 0 && !r.done { + if err := r.poll(); err != nil { + // An interrupted command is the reader leaving, not a failure. + if r.ctx.Err() != nil { + r.done = true + break + } + return nil, fmt.Errorf("failed reading %v: %w", r.target, plainRefusal(err)) + } + } + if len(r.buf) == 0 { + return nil, nil + } + rec := r.buf[0] + r.buf = r.buf[1:] + r.seen++ + return rec, nil +} + +func (r *streamReader) poll() error { + out, err := r.cl.poll(r.ctx, r.target, r.from, r.start, r.pageSize, r.topics, r.follow) + if err != nil { + return err + } + // The first poll resolved the start; from here the cursor is an offset. + r.start = nil + r.from = out.GetNextOffset() + if err := r.cl.decodeRecords(out.GetRecords()); err != nil { + return fmt.Errorf("decoding records: %w", err) + } + r.buf = append(r.buf, out.GetRecords()...) + caughtUp := out.GetNextOffset() >= out.GetHeadOffset() + if caughtUp && (out.GetClosed() || !r.follow) { + r.done = true + } + return nil +} + +type streamRecordRow struct { + Offset int64 + Kind string + Topic string + Producer string + Attempt int64 + Sequence int64 + Body string +} + +// streamRowIter adapts the reader to the printer's streaming table. +type streamRowIter struct { + cctx *CommandContext + reader *streamReader +} + +func (i *streamRowIter) Next() (any, error) { + rec, err := i.reader.next() + if rec == nil || err != nil { + return nil, err + } + body, err := payloadText(rec.GetBody()) + if err != nil { + return nil, err + } + return streamRecordRow{ + Offset: rec.GetOffset(), + Kind: recordKindText(rec.GetKind()), + Topic: rec.GetTopic(), + Producer: rec.GetProducerId(), + Attempt: rec.GetAttempt(), + Sequence: rec.GetSequence(), + Body: body, + }, nil +} + +// recordKindText is the kind in the shorthand the API's enums print with, the +// way every other enum in a table reads. +func recordKindText(kind streamapi.StreamRecordKind) string { + if kind == streamapi.STREAM_RECORD_KIND_UNSPECIFIED { + kind = streamapi.STREAM_RECORD_KIND_DATA + } + return kind.String() +} + +func (c *TemporalStreamReadCommand) run(cctx *CommandContext, _ []string) error { + target, err := c.target() + if err != nil { + return err + } + start, err := c.startPosition() + if err != nil { + return err + } + if c.PageSize <= 0 { + return fmt.Errorf("--page-size must be positive") + } + cl, err := dialStreamClient(cctx, &c.Parent.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + reader := &streamReader{ + ctx: cctx, + cl: cl, + target: target, + start: start, + topics: c.Topic, + follow: c.Follow, + pageSize: int32(c.PageSize), + limit: c.Limit, + } + if cctx.JSONOutput { + // This is a listing command subject to json vs jsonl rules + cctx.Printer.StartList() + defer cctx.Printer.EndList() + for { + rec, err := reader.next() + if err != nil { + return err + } + if rec == nil { + return nil + } + if err := cctx.Printer.PrintStructured(rec, printer.StructuredOptions{}); err != nil { + return err + } + } + } + return cctx.Printer.PrintStructuredTableIter( + reflect.TypeOf(streamRecordRow{}), + &streamRowIter{cctx: cctx, reader: reader}, + printer.StructuredOptions{ + Table: &printer.TableOptions{ + // Streaming rows cannot be measured first, so the columns + // before the body take fixed widths. + FieldWidths: map[string]int{ + "Offset": 8, + "Kind": 8, + "Topic": 16, + "Producer": 20, + "Attempt": 8, + "Sequence": 10, + }, + }, + }, + ) +} + +// inputPayloads takes the records from --input and --input-file, or one per +// line of stdin when neither is given. +func (c *TemporalStreamAppendCommand) inputPayloads( + cctx *CommandContext, +) ([]*commonpb.Payload, error) { + if len(c.Input) == 0 && len(c.InputFile) == 0 { + lines, err := readStdinLines(cctx) + if err != nil { + return nil, err + } + c.Input = lines + } + if len(c.Input) == 0 && len(c.InputFile) == 0 { + return nil, nil + } + payloads, err := c.buildRawInputPayloads() + if err != nil { + return nil, err + } + return payloads.GetPayloads(), nil +} + +func readStdinLines(cctx *CommandContext) ([]string, error) { + // A terminal with nothing piped in would sit waiting for input the user + // did not mean to type. + if f, ok := cctx.Options.Stdin.(*os.File); ok && isatty.IsTerminal(f.Fd()) { + return nil, nil + } + scanner := bufio.NewScanner(cctx.Options.Stdin) + scanner.Buffer(make([]byte, 0, 64*1024), 16<<20) + var lines []string + for scanner.Scan() { + if line := strings.TrimSpace(scanner.Text()); line != "" { + lines = append(lines, line) + } + } + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("failed reading stdin: %w", err) + } + return lines, nil +} + +func (c *TemporalStreamAppendCommand) run(cctx *CommandContext, _ []string) error { + target, err := c.target() + if err != nil { + return err + } + var expectedOffset *int64 + if c.Command.Flags().Changed("expected-offset") { + if target.streamID == "" { + return fmt.Errorf("--expected-offset applies to a standalone stream") + } + if c.ExpectedOffset < 0 { + return fmt.Errorf("--expected-offset cannot be negative") + } + v := int64(c.ExpectedOffset) + expectedOffset = &v + } + if c.Attempt < 0 || c.Sequence < 0 { + return fmt.Errorf("--attempt and --sequence cannot be negative") + } + sequenceSet := c.Command.Flags().Changed("sequence") + if sequenceSet && c.ProducerId == "" { + return fmt.Errorf("--sequence needs --producer-id") + } + payloads, err := c.inputPayloads(cctx) + if err != nil { + return err + } + if len(payloads) == 0 && !c.Finish { + return fmt.Errorf( + "no records to append: pass --input, --input-file, lines on stdin, or --finish") + } + + records := make([]*streampb.StreamRecord, 0, len(payloads)+1) + for i, p := range payloads { + rec := &streampb.StreamRecord{ + Body: p, + Topic: c.Topic, + Kind: streamapi.STREAM_RECORD_KIND_DATA, + ProducerId: c.ProducerId, + Attempt: int64(c.Attempt), + } + if sequenceSet { + rec.Sequence = int64(c.Sequence + i) + } + records = append(records, rec) + } + if c.Finish { + rec := &streampb.StreamRecord{ + Topic: c.Topic, + Kind: streamapi.STREAM_RECORD_KIND_FINISH, + ProducerId: c.ProducerId, + Attempt: int64(c.Attempt), + } + if sequenceSet { + rec.Sequence = int64(c.Sequence + len(payloads)) + } + records = append(records, rec) + } + + cl, err := dialStreamClient(cctx, &c.Parent.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + if err := cl.encodeRecords(records); err != nil { + return fmt.Errorf("failed encoding records: %w", err) + } + var sequence int64 + if sequenceSet { + sequence = int64(c.Sequence) + } + out, err := cl.append(cctx, target, records, c.ProducerId, sequence, expectedOffset) + if err != nil { + return fmt.Errorf("failed appending to %v: %w", target, plainRefusal(err)) + } + if cctx.JSONOutput { + return cctx.Printer.PrintStructured(out, printer.StructuredOptions{}) + } + return cctx.Printer.PrintStructured(struct { + FirstOffset int64 + NextOffset int64 + Count int64 + Deduplicated bool + }{ + FirstOffset: out.GetFirstOffset(), + NextOffset: out.GetNextOffset(), + Count: out.GetCount(), + Deduplicated: out.GetDeduplicated(), + }, printer.StructuredOptions{}) +} + +func (c *TemporalStreamCloseCommand) run(cctx *CommandContext, _ []string) error { + cl, err := dialStreamClient(cctx, &c.Parent.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + var reason *commonpb.Payload + if c.Reason != "" { + reason, err = converter.GetDefaultDataConverter().ToPayload(c.Reason) + if err != nil { + return fmt.Errorf("failed converting reason: %w", err) + } + if cl.PayloadCodec != nil { + encoded, err := cl.PayloadCodec.Encode([]*commonpb.Payload{reason}) + if err != nil { + return fmt.Errorf("failed encoding reason: %w", err) + } + reason = encoded[0] + } + } + _, err = cl.svc.CloseStream(cctx, &streampb.CloseStreamRequest{ + FrontendRequest: &streampb.CloseStreamInput{ + Namespace: cl.namespace, + StreamId: c.StreamId, + Reason: reason, + }, + }) + if err != nil { + return fmt.Errorf("failed closing stream %q: %w", c.StreamId, err) + } + cctx.Printer.Printlnf("Closed stream %v", c.StreamId) + return nil +} + +func (c *TemporalStreamTruncateCommand) run(cctx *CommandContext, _ []string) error { + if c.To < 0 { + return fmt.Errorf("--to cannot be negative") + } + cl, err := dialStreamClient(cctx, &c.Parent.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + _, err = cl.svc.TruncateStream(cctx, &streampb.TruncateStreamRequest{ + FrontendRequest: &streampb.TruncateStreamInput{ + Namespace: cl.namespace, + StreamId: c.StreamId, + NewBaseOffset: int64(c.To), + }, + }) + if err != nil { + return fmt.Errorf("failed truncating stream %q: %w", c.StreamId, err) + } + cctx.Printer.Printlnf("Truncated stream %v to offset %v", c.StreamId, c.To) + return nil +} + +func (c *TemporalStreamDeleteCommand) run(cctx *CommandContext, _ []string) error { + cl, err := dialStreamClient(cctx, &c.Parent.ClientOptions) + if err != nil { + return err + } + defer cl.Close() + + _, err = cl.svc.DeleteStream(cctx, &streampb.DeleteStreamRequest{ + FrontendRequest: &streampb.DeleteStreamInput{ + Namespace: cl.namespace, + StreamId: c.StreamId, + Force: c.Force, + }, + }) + if err != nil { + return fmt.Errorf("failed deleting stream %q: %w", c.StreamId, err) + } + cctx.Printer.Printlnf("Deleted stream %v", c.StreamId) + return nil +} diff --git a/internal/temporalcli/commands.stream_test.go b/internal/temporalcli/commands.stream_test.go new file mode 100644 index 000000000..7700cac26 --- /dev/null +++ b/internal/temporalcli/commands.stream_test.go @@ -0,0 +1,477 @@ +package temporalcli_test + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/temporalio/cli/internal/temporalcli" + notificationpb "go.temporal.io/api/notification/v1" + streamapi "go.temporal.io/api/stream/v1" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/workflow" + streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1" +) + +// fakeStreamService answers every describe with one canned state and records +// the owners the CLI named, so the output is checked without a server that +// carries streams. +type fakeStreamService struct { + streampb.UnimplementedStreamServiceServer + + mu sync.Mutex + state *streampb.StreamState + owned []*streampb.DescribeWorkflowStreamInput +} + +func (f *fakeStreamService) DescribeStream( + context.Context, *streampb.DescribeStreamRequest, +) (*streampb.DescribeStreamResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + return &streampb.DescribeStreamResponse{ + FrontendResponse: &streampb.DescribeStreamOutput{State: f.state}, + }, nil +} + +func (f *fakeStreamService) DescribeWorkflowStream( + _ context.Context, req *streampb.DescribeWorkflowStreamRequest, +) (*streampb.DescribeWorkflowStreamResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.owned = append(f.owned, req.GetFrontendRequest()) + return &streampb.DescribeWorkflowStreamResponse{ + FrontendResponse: &streampb.DescribeStreamOutput{State: f.state}, + }, nil +} + +func TestStream_DescribeChannel(t *testing.T) { + st := &fakeStreamService{state: &streampb.StreamState{HeadOffset: 2}} + addr := startFakeServices(t, &fakeChannelService{}, st) + h := NewCommandHarness(t) + // channelSection is the output from the channel card on, so the owner + // asserted on is the channel's and not the stream card's. + channelSection := func(args ...string) string { + res := h.Execute(append([]string{"stream", "describe", "--address", addr}, args...)...) + require.NoError(t, res.Err) + out := res.Stdout.String() + h.ContainsOnSameLine(out, "HeadOffset", "2") + _, section, found := strings.Cut(out, "Notification Channel:") + require.True(t, found, "no channel card in %q", out) + return section + } + + section := channelSection("--stream-id", "scores") + h.ContainsOnSameLine(section, "Channel", "stream/scores") + h.ContainsOnSameLine(section, "Kind", "Independent") + assert.NotContains(t, section, "LinkedTo") + + // The owner's default stream has the name the server resolves it to. + section = channelSection("--workflow-id", "wf-1") + h.ContainsOnSameLine(section, "Channel", "stream/output") + h.ContainsOnSameLine(section, "Kind", "Linked") + h.ContainsOnSameLine(section, "LinkedTo", "workflow wf-1") + assert.NotContains(t, section, "run ") + + section = channelSection("--workflow-id", "wf-1", "--run-id", "run-1", "--name", "scores") + h.ContainsOnSameLine(section, "Channel", "stream/scores") + h.ContainsOnSameLine(section, "Kind", "Linked") + h.ContainsOnSameLine(section, "LinkedTo", "workflow wf-1 (run run-1)") + + // An activity the workflow scheduled notifies the workflow, under a name + // that carries the activity ID. + section = channelSection("--workflow-id", "wf-1", "--activity-id", "act-1", "--name", "scores") + h.ContainsOnSameLine(section, "Channel", "stream/act-1/scores") + h.ContainsOnSameLine(section, "Kind", "Linked") + h.ContainsOnSameLine(section, "LinkedTo", "workflow wf-1") + + // A standalone activity holds linked channels of its own, so its stream + // notifies one named like a workflow's and linked to the activity. + section = channelSection("--activity-id", "act-1", "--run-id", "run-2", "--name", "scores") + h.ContainsOnSameLine(section, "Channel", "stream/scores") + h.ContainsOnSameLine(section, "Kind", "Linked") + h.ContainsOnSameLine(section, "LinkedTo", "activity act-1 (run run-2)") + + // The derivation adds no call: each describe reached the service once. + st.mu.Lock() + defer st.mu.Unlock() + require.Len(t, st.owned, 4) + assert.Equal(t, "", st.owned[0].GetStreamName(), "the server resolves the default name") + assert.Equal(t, streampb.STREAM_OWNER_KIND_ACTIVITY, st.owned[3].GetOwner().GetKind()) +} + +func (s *SharedServerSuite) createStream(id string, args ...string) { + res := s.Execute(append([]string{ + "stream", "create", "--address", s.Address(), "--stream-id", id, + }, args...)...) + s.NoError(res.Err) + s.Contains(res.Stdout.String(), id) +} + +// readStreamJSON runs a read with JSON output and decodes every record. +func (s *SharedServerSuite) readStreamJSON(args ...string) []*streampb.StreamRecord { + res := s.Execute(append([]string{ + "stream", "read", "--address", s.Address(), "-o", "json", + }, args...)...) + s.NoError(res.Err) + var raw []json.RawMessage + s.NoError(json.Unmarshal(res.Stdout.Bytes(), &raw)) + records := make([]*streampb.StreamRecord, len(raw)) + for i, r := range raw { + records[i] = &streampb.StreamRecord{} + s.NoError(temporalcli.UnmarshalProtoJSONWithOptions(r, records[i], true)) + } + return records +} + +func (s *SharedServerSuite) TestStream_StandaloneLifecycle() { + id := "stream-" + uuid.NewString() + s.createStream(id, "--retention", "1h", "--max-bytes", "1048576") + + // The same lifecycle again is a retry; a different one is refused. + res := s.Execute( + "stream", "create", "--address", s.Address(), "--stream-id", id, + "--retention", "1h", "--max-bytes", "1048576", + ) + s.ErrorContains(res.Err, "already exists") + res = s.Execute( + "stream", "create", "--address", s.Address(), "--stream-id", id, + "--retention", "1h", "--max-bytes", "2048", + ) + s.ErrorContains(res.Err, "different lifecycle") + s.NotContains(res.Err.Error(), "STREAM_POLICY_MISMATCH") + + appendArgs := []string{ + "stream", "append", "--address", s.Address(), "--stream-id", id, + "--topic", "scores", "--producer-id", "p1", "--attempt", "1", "--sequence", "1", + "--input", `{"home": 1}`, "--input", `{"home": 2}`, + } + res = s.Execute(appendArgs...) + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "FirstOffset", "0") + s.ContainsOnSameLine(res.Stdout.String(), "NextOffset", "2") + s.ContainsOnSameLine(res.Stdout.String(), "Deduplicated", "false") + + // The same producer retrying the same sequence appends nothing. + res = s.Execute(appendArgs...) + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "Deduplicated", "true") + + // Different content at that sequence is a conflict, and an older sequence + // is stale; both are refused in plain words, without the token. + res = s.Execute( + "stream", "append", "--address", s.Address(), "--stream-id", id, + "--producer-id", "p1", "--sequence", "1", "--input", `{"home": 9}`, + ) + s.ErrorContains(res.Err, "different content at this sequence") + s.NotContains(res.Err.Error(), "STREAM_PRODUCER_CONFLICT") + res = s.Execute( + "stream", "append", "--address", s.Address(), "--stream-id", id, + "--producer-id", "p1", "--sequence", "0", "--input", `{"home": 9}`, + ) + s.ErrorContains(res.Err, "below the producer's latest") + s.NotContains(res.Err.Error(), "STREAM_PRODUCER_STALE_SEQUENCE") + + // One record per stdin line, then the producer's FINISH. + s.Stdin.WriteString("\"third\"\n\n\"fourth\"\n") + res = s.Execute( + "stream", "append", "--address", s.Address(), "--stream-id", id, + "--producer-id", "p2", "--finish", + ) + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "FirstOffset", "2") + s.ContainsOnSameLine(res.Stdout.String(), "Count", "3") + + // Nothing to append is an error rather than an empty batch. + res = s.Execute("stream", "append", "--address", s.Address(), "--stream-id", id) + s.ErrorContains(res.Err, "no records to append") + + // Read from the beginning as a table. + res = s.Execute("stream", "read", "--address", s.Address(), "--stream-id", id) + s.NoError(res.Err) + out := res.Stdout.String() + s.ContainsOnSameLine(out, "Offset", "Kind", "Topic", "Producer", "Attempt", "Sequence", "Body") + s.ContainsOnSameLine(out, "0", "Data", "scores", "p1", "1", "1", `{"home":1}`) + s.ContainsOnSameLine(out, "1", "Data", "scores", "p1", "1", "2", `{"home":2}`) + s.ContainsOnSameLine(out, "2", "Data", "p2", `"third"`) + s.ContainsOnSameLine(out, "3", "Data", "p2", `"fourth"`) + s.ContainsOnSameLine(out, "4", "Finish", "p2") + + // The same records as JSON, with the body in shorthand. + records := s.readStreamJSON("--stream-id", id) + s.Len(records, 5) + s.Equal(int64(3), records[3].GetOffset()) + s.Equal(`"fourth"`, string(records[3].GetBody().GetData())) + s.Equal("p2", records[4].GetProducerId()) + s.Equal(streamapi.STREAM_RECORD_KIND_FINISH, records[4].GetKind()) + + // Start positions and filters. + records = s.readStreamJSON("--stream-id", id, "--last", "2") + s.Len(records, 2) + s.Equal(int64(3), records[0].GetOffset()) + records = s.readStreamJSON("--stream-id", id, "--from-offset", "4") + s.Len(records, 1) + s.Equal(int64(4), records[0].GetOffset()) + records = s.readStreamJSON("--stream-id", id, "--topic", "scores") + s.Len(records, 2) + records = s.readStreamJSON("--stream-id", id, "--limit", "3", "--page-size", "2") + s.Len(records, 3) + records = s.readStreamJSON("--stream-id", id, "--from-tail") + s.Empty(records) + + // Describe shows the frontier and both producers. + res = s.Execute("stream", "describe", "--address", s.Address(), "--stream-id", id) + s.NoError(res.Err) + out = res.Stdout.String() + s.ContainsOnSameLine(out, "StreamId", id) + s.ContainsOnSameLine(out, "HeadOffset", "5") + s.ContainsOnSameLine(out, "Retention", "1h") + s.ContainsOnSameLine(out, "MaxBytes", "1048576") + s.ContainsOnSameLine(out, "HeldBytes") + s.ContainsOnSameLine(out, "Producers", "2") + s.ContainsOnSameLine(out, "p1", "1", "0", "2", "false") + // A FINISH record tells readers the producer is done; only FinishWriting + // fences it on the stream. + s.ContainsOnSameLine(out, "p2", "0", "2", "3", "false") + s.ContainsOnSameLine(out, "Closed", "false") + s.ContainsOnSameLine(out, "Consumers", "0") + + res = s.Execute("stream", "describe", "--address", s.Address(), "--stream-id", id, "-o", "json") + s.NoError(res.Err) + var state streampb.StreamState + s.NoError(temporalcli.UnmarshalProtoJSONWithOptions(res.Stdout.Bytes(), &state, true)) + s.Equal(int64(5), state.GetHeadOffset()) + s.Equal(int64(0), state.GetBaseOffset()) + s.False(state.GetClosed()) + s.Equal(int64(3), state.GetProducers()["p2"].GetCount()) + s.Equal(int64(1048576), state.GetLifecycle().GetMaxBytes()) + s.Positive(state.GetHeldBytes()) + s.Equal(state.GetAppendedBytes(), state.GetHeldBytes()) + + // Truncation moves the floor; the beginning is now the oldest record left, + // and the reclaimed batch no longer counts as held. + res = s.Execute("stream", "truncate", "--address", s.Address(), "--stream-id", id, "--to", "2") + s.NoError(res.Err) + records = s.readStreamJSON("--stream-id", id) + s.Len(records, 3) + s.Equal(int64(2), records[0].GetOffset()) + res = s.Execute("stream", "describe", "--address", s.Address(), "--stream-id", id, "-o", "json") + s.NoError(res.Err) + var truncated streampb.StreamState + s.NoError(temporalcli.UnmarshalProtoJSONWithOptions(res.Stdout.Bytes(), &truncated, true)) + s.Equal(int64(2), truncated.GetBaseOffset()) + s.Less(truncated.GetHeldBytes(), truncated.GetAppendedBytes()) + res = s.Execute( + "stream", "read", "--address", s.Address(), "--stream-id", id, "--from-offset", "0", + ) + s.ErrorContains(res.Err, "below the stream's floor") + s.NotContains(res.Err.Error(), "STREAM_CURSOR_BELOW_FLOOR") + + // Closing seals the stream but keeps it readable. + res = s.Execute("stream", "close", "--address", s.Address(), "--stream-id", id, "--reason", "done") + s.NoError(res.Err) + res = s.Execute("stream", "describe", "--address", s.Address(), "--stream-id", id) + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "Closed", "true") + s.ContainsOnSameLine(res.Stdout.String(), "CloseReason", `"done"`) + res = s.Execute( + "stream", "append", "--address", s.Address(), "--stream-id", id, "--input", `"late"`, + ) + s.ErrorContains(res.Err, "the stream is closed") + s.NotContains(res.Err.Error(), "STREAM_CLOSED") + s.Len(s.readStreamJSON("--stream-id", id), 3) + + // Deleting takes the records with it. + res = s.Execute("stream", "delete", "--address", s.Address(), "--stream-id", id) + s.NoError(res.Err) + res = s.Execute("stream", "describe", "--address", s.Address(), "--stream-id", id) + s.Error(res.Err) +} + +func (s *SharedServerSuite) TestStream_ReadFollow() { + id := "stream-" + uuid.NewString() + s.createStream(id) + res := s.Execute( + "stream", "append", "--address", s.Address(), "--stream-id", id, "--input", `"first"`, + ) + s.NoError(res.Err) + + done := make(chan *CommandResult, 1) + go func() { + done <- s.Execute("stream", "read", "--address", s.Address(), "--stream-id", id, "--follow") + }() + + // The follow is parked on the frontier by the time this lands. + time.Sleep(time.Second) + res = s.Execute( + "stream", "append", "--address", s.Address(), "--stream-id", id, "--input", `"second"`, + ) + s.NoError(res.Err) + res = s.Execute("stream", "close", "--address", s.Address(), "--stream-id", id) + s.NoError(res.Err) + + select { + case res = <-done: + case <-time.After(60 * time.Second): + s.Fail("follow did not end after the stream closed") + } + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "0", "Data", `"first"`) + s.ContainsOnSameLine(res.Stdout.String(), "1", "Data", `"second"`) +} + +func (s *SharedServerSuite) TestStream_OwnedByWorkflow() { + s.Worker().OnDevWorkflow(func(ctx workflow.Context, a any) (any, error) { + workflow.GetSignalChannel(ctx, "finish").Receive(ctx, nil) + return nil, nil + }) + run, err := s.Client.ExecuteWorkflow( + s.Context, + client.StartWorkflowOptions{TaskQueue: s.Worker().Options.TaskQueue}, + DevWorkflow, + "ignored", + ) + s.NoError(err) + + // An owned stream comes into being on its first append. + res := s.Execute( + "stream", "append", "--address", s.Address(), + "--workflow-id", run.GetID(), "--name", "scores", + "--producer-id", "outside", "--input", `{"home": 1}`, "--input", `{"home": 2}`, + ) + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "NextOffset", "2") + + res = s.Execute( + "stream", "read", "--address", s.Address(), + "--workflow-id", run.GetID(), "--run-id", run.GetRunID(), "--name", "scores", + ) + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "0", "Data", "outside", `{"home":1}`) + s.ContainsOnSameLine(res.Stdout.String(), "1", "Data", "outside", `{"home":2}`) + + res = s.Execute( + "stream", "describe", "--address", s.Address(), "--workflow-id", run.GetID(), "--name", "scores", + ) + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "WorkflowId", run.GetID()) + s.ContainsOnSameLine(res.Stdout.String(), "Name", "scores") + s.ContainsOnSameLine(res.Stdout.String(), "HeadOffset", "2") + + // The owner's default stream is a different, still empty, stream. + s.Empty(s.readStreamJSON("--workflow-id", run.GetID())) + + s.NoError(s.Client.SignalWorkflow(s.Context, run.GetID(), "", "finish", nil)) + s.NoError(run.Get(s.Context, nil)) +} + +func (s *SharedServerSuite) TestStream_List() { + prefix := "list-" + uuid.NewString()[:8] + s.createStream(prefix + "-a") + s.createStream(prefix + "-b") + other := "other-" + uuid.NewString() + s.createStream(other) + query := fmt.Sprintf(`WorkflowId STARTS_WITH "%s"`, prefix) + + var out string + s.EventuallyWithT(func(t *assert.CollectT) { + res := s.Execute("stream", "list", "--address", s.Address(), "--query", query) + assert.NoError(t, res.Err) + out = res.Stdout.String() + assert.Contains(t, out, prefix+"-a") + assert.Contains(t, out, prefix+"-b") + }, 10*time.Second, 200*time.Millisecond) + s.ContainsOnSameLine(out, "StreamId", "RunId") + s.NotContains(out, other) + + res := s.Execute( + "stream", "list", "--address", s.Address(), "--query", query, "--limit", "1", "-o", "json", + ) + s.NoError(res.Err) + var entries []json.RawMessage + s.NoError(json.Unmarshal(res.Stdout.Bytes(), &entries)) + s.Len(entries, 1) + var entry streampb.StreamListEntry + s.NoError(temporalcli.UnmarshalProtoJSONWithOptions(entries[0], &entry, true)) + s.Contains(entry.GetStreamId(), prefix) + s.NotEmpty(entry.GetRunId()) +} + +func (s *SharedServerSuite) TestStream_StandaloneNotifiesItsChannel() { + id := "stream-" + uuid.NewString() + s.createStream(id) + ch := "stream/" + id + poll := func(after int64) *notificationpb.Notification { + res := s.Execute("channel", "poll", "--address", s.Address(), "-c", ch, + "--after-counter", strconv.FormatInt(after, 10), "--wait", "10s", "-o", "jsonl") + s.NoError(res.Err) + raw := decodeJSONValues(s.T(), res.Stdout.String()) + s.NotEmpty(raw) + var n notificationpb.Notification + s.NoError(temporalcli.UnmarshalProtoJSONWithOptions(raw[len(raw)-1], &n, true)) + return &n + } + + res := s.Execute("stream", "describe", "--address", s.Address(), "--stream-id", id) + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "Channel", ch) + s.ContainsOnSameLine(res.Stdout.String(), "Kind", "Independent") + + res = s.Execute("stream", "append", "--address", s.Address(), "--stream-id", id, + "--input", `{"home": 1}`) + s.NoError(res.Err) + // A standalone stream's notify reaches its channel through a task of its + // own, so the poll waits for it. + n := poll(0) + s.Equal(ch, n.GetChannel()) + s.GreaterOrEqual(n.GetCounter(), int64(1)) + s.Equal(id+":1", string(n.GetPosition())) + s.Nil(n.GetMetadata()["closed"]) + + res = s.Execute("stream", "close", "--address", s.Address(), "--stream-id", id) + s.NoError(res.Err) + closed := poll(n.GetCounter()) + s.Greater(closed.GetCounter(), n.GetCounter()) + s.Equal(id+":1", string(closed.GetPosition())) + s.Equal("true", string(closed.GetMetadata()["closed"].GetData())) + + res = s.Execute("channel", "describe", "--address", s.Address(), "-c", ch) + s.NoError(res.Err) + s.ContainsOnSameLine(res.Stdout.String(), "Kind", "Independent") + s.ContainsOnSameLine(res.Stdout.String(), "LatestCounter", + strconv.FormatInt(closed.GetCounter(), 10)) +} + +func (s *SharedServerSuite) TestStream_ReferenceValidation() { + res := s.Execute( + "stream", "read", "--address", s.Address(), "--stream-id", "a", "--workflow-id", "b", + ) + s.ErrorContains(res.Err, "cannot set --stream-id") + res = s.Execute("stream", "read", "--address", s.Address()) + s.ErrorContains(res.Err, "must set either --stream-id") + res = s.Execute("stream", "read", "--address", s.Address(), "--stream-id", "a", "--name", "n") + s.ErrorContains(res.Err, "--run-id and --name") + res = s.Execute( + "stream", "read", "--address", s.Address(), "--stream-id", "a", "--from-tail", "--last", "2", + ) + s.ErrorContains(res.Err, "at most one") + res = s.Execute( + "stream", "append", "--address", s.Address(), "--workflow-id", "w", + "--expected-offset", "3", "--input", "1", + ) + s.ErrorContains(res.Err, "--expected-offset applies to a standalone stream") + res = s.Execute("stream", "truncate", "--address", s.Address(), "--to", "1") + s.ErrorContains(res.Err, "stream-id") + res = s.Execute( + "stream", "create", "--address", s.Address(), "--stream-id", "a", "--max-bytes", "-1", + ) + s.ErrorContains(res.Err, "cannot be negative") +} diff --git a/internal/temporalcli/commands.yaml b/internal/temporalcli/commands.yaml index 18b1020fe..30ad1c46d 100644 --- a/internal/temporalcli/commands.yaml +++ b/internal/temporalcli/commands.yaml @@ -3619,6 +3619,14 @@ commands: --ui-port 3000 ``` + Streams are enabled on the development server. Turn them off with a + dynamic configuration value: + + ``` + temporal server start-dev \ + --dynamic-config-value 'stream.enabled=false' + ``` + Callbacks to `127.0.0.1` and `localhost` on any port are allowed over plain HTTP, so a local receiver can listen on a notification channel. Replace the list with a dynamic configuration value: @@ -3710,6 +3718,319 @@ commands: attribute type, which is one of the following: Text, Keyword, Int, Double, Bool, Datetime, KeywordList. + - name: temporal stream + summary: Read and manage Streams + description: | + A Stream is an append-only log the Temporal Service carries next to + Workflow Executions. A standalone Stream has an ID of its own. An owned + Stream lives inside a Workflow Execution or an Activity and is addressed + by its owner and a Stream name: + + ``` + temporal stream [command] [options] + ``` + + For example: + + ``` + temporal stream read \ + --workflow-id YourWorkflowId + ``` + + The Service must enable Streams for the Namespace through the + `stream.enabled` dynamic configuration value. The development server + enables them. + option-sets: + - client + docs: + description-header: >- + Temporal Stream commands list, describe, read, append to, close, + truncate and delete the Streams a Temporal Service carries. + keywords: + - cli reference + - command-line-interface-cli + - stream + - stream append + - stream close + - stream create + - stream delete + - stream describe + - stream list + - stream read + - stream truncate + - streams + - temporal cli + tags: + - Temporal CLI + - Streams + + - name: temporal stream append + summary: Append records to a Stream + description: | + Append one or more records to a Stream. Each `--input` is one record. + Without `--input` or `--input-file`, one record is read per line from + stdin: + + ``` + temporal stream append \ + --stream-id YourStreamId \ + --input '{"score": 1}' \ + --input '{"score": 2}' + ``` + + Append to a Stream a Workflow Execution owns, naming the owner instead: + + ``` + temporal stream append \ + --workflow-id YourWorkflowId \ + --name scores \ + --input '{"score": 1}' + ``` + + Set `--producer-id` and `--sequence` so a retried append is deduplicated + rather than appended twice. Pass `--finish` to end the producer's writes + on the topic with a `FINISH` record after the values. + options: + - name: topic + type: string + description: Topic label stored on every record. + - name: producer-id + type: string + description: | + Producer identity stored on every record. + Together with --sequence it deduplicates a retried append. + - name: attempt + type: int + description: Producer attempt stored on every record. + - name: sequence + type: int + description: | + Sequence of the first record within the producer's attempt. + Records that follow take the next numbers. + - name: expected-offset + type: int + description: | + Refuse the append unless the Stream's next offset equals this value. + Only applies to a standalone Stream. + - name: finish + type: bool + description: Append a `FINISH` record after the values. + option-sets: + - stream-reference + - payload-input + + - name: temporal stream close + summary: Close a Stream + description: | + Close a standalone Stream. A closed Stream accepts no more records and + stays readable until its retention passes: + + ``` + temporal stream close \ + --stream-id YourStreamId \ + --reason "done" + ``` + options: + - name: reason + type: string + description: Reason recorded on the Stream. + option-sets: + - stream-id + + - name: temporal stream create + summary: Create a standalone Stream + description: | + Create a standalone Stream with an ID of its own: + + ``` + temporal stream create \ + --stream-id YourStreamId + ``` + + Records older than `--retention` are reclaimed, and a closed Stream is + deleted once it passes. Without it the Namespace's retention applies. + Creating a Stream that already exists with the same lifecycle is an + idempotent retry; a different lifecycle is refused. + options: + - name: retention + type: duration + description: | + How long a record stays readable, and how long a closed Stream + stays readable before it is deleted. + Cannot be longer than the Namespace's retention. + - name: max-items + type: int + description: | + Most records the Stream holds before the oldest are reclaimed. + Default is zero (unbounded). + - name: max-bytes + type: int + description: | + Most bytes the Stream holds. An append past it is refused until + records are reclaimed. Default is zero (unbounded). + option-sets: + - stream-id + + - name: temporal stream delete + summary: Remove a Stream + description: | + Delete a standalone Stream and every record it holds: + + ``` + temporal stream delete \ + --stream-id YourStreamId + ``` + + The call is refused while a Workflow Execution consumes the Stream, + because its replay depends on the records. Pass `--force` to delete it + anyway. + options: + - name: force + type: bool + description: Delete even while a Workflow Execution consumes the Stream. + option-sets: + - stream-id + + - name: temporal stream describe + summary: Show a Stream's state + description: | + Show a Stream's frontier, floor, close state, retention, producers and + consumers: + + ``` + temporal stream describe \ + --stream-id YourStreamId + ``` + + Describe a Stream a Workflow Execution owns: + + ``` + temporal stream describe \ + --workflow-id YourWorkflowId \ + --name scores + ``` + + The output also names the notification channel the Stream notifies on + each append and on its close. The name follows from the Stream alone, + so no call to the Service is needed. A Stream a Workflow Execution + owns notifies `stream/NAME`, linked to the owner. A Stream of an + Activity the Workflow scheduled notifies `stream/ACTIVITY_ID/NAME`, + linked to the owner too. A standalone Stream notifies the independent + channel `stream/STREAM_ID`. Pass the channel, and the owner of a + linked one, to `temporal channel poll` to follow the Stream: + + ``` + temporal channel poll \ + --channel stream/scores \ + --workflow-id YourWorkflowId + ``` + option-sets: + - stream-reference + + - name: temporal stream list + summary: Show standalone Streams + description: | + List the standalone Streams in a Namespace. The optional `--query` + limits the output to Streams matching a List Filter, where `WorkflowId` + holds the Stream ID: + + ``` + temporal stream list \ + --query 'WorkflowId STARTS_WITH "scores-"' + ``` + + Streams a Workflow Execution owns are not listed here. Describe or read + them through their owner. + options: + - name: query + short: q + type: string + description: Content for an SQL-like `QUERY` List Filter. + - name: limit + type: int + description: Maximum number of Streams to display. + - name: page-size + type: int + description: Maximum number of Streams to fetch at a time from the server. + default: 100 + + - name: temporal stream read + summary: Read a Stream's records + description: | + Read a Stream from the beginning and stop once caught up: + + ``` + temporal stream read \ + --stream-id YourStreamId + ``` + + Read a Stream a Workflow Execution owns and keep following it as records + land: + + ``` + temporal stream read \ + --workflow-id YourWorkflowId \ + --name scores \ + --follow + ``` + + Start at a cursor with `--from-offset`, at the tail with `--from-tail`, + or at the newest records with `--last`. Each record prints with its + offset, kind, topic, producer, attempt, sequence and body. With + `--output json` each record is one JSON object. + options: + - name: from-offset + type: int + description: | + Offset to start at, inclusive. + Refused when below the Stream's floor. + - name: from-tail + type: bool + description: Start at the tail and show only records appended from now on. + - name: last + type: int + description: Start at the newest records, showing at most this many first. + - name: follow + short: f + type: bool + description: Keep reading as records land, until the Stream closes. + - name: topic + type: string[] + description: | + Show only records on this topic. + Can be passed multiple times. + - name: limit + type: int + description: Maximum number of records to display. + - name: page-size + type: int + description: Maximum number of records to fetch at a time from the server. + default: 100 + option-sets: + - stream-reference + + - name: temporal stream truncate + summary: Drop the oldest records of a Stream + description: | + Move the readable floor of a standalone Stream up to an offset. Records + below it are gone: + + ``` + temporal stream truncate \ + --stream-id YourStreamId \ + --to 100 + ``` + + The call is refused while a Workflow Execution consumes records below + the new floor. + options: + - name: to + type: int + description: New floor. Records below this offset are dropped. + required: true + option-sets: + - stream-id + - name: temporal task-queue summary: Manage Task Queues description: | @@ -5665,6 +5986,46 @@ option-sets: Run ID of the linked execution. Defaults to the current run. Requires --workflow-id or --activity-id. + - name: stream-id + options: + - name: stream-id + type: string + short: s + description: Stream ID of a standalone Stream. + required: true + + - name: stream-reference + options: + - name: stream-id + type: string + short: s + description: | + Stream ID of a standalone Stream. + You must set either --stream-id or an owner. + - name: workflow-id + type: string + short: w + description: | + Workflow ID of the Workflow Execution owning the Stream. + Cannot use with --stream-id. + - name: run-id + type: string + short: r + description: | + Run ID of the owner. + Defaults to the current run. + - name: activity-id + type: string + description: | + Activity ID of the Activity owning the Stream. + With --workflow-id it names an Activity the Workflow scheduled. + On its own it names a standalone Activity. + - name: name + type: string + description: | + Name of the Stream within its owner. + Defaults to the owner's default Stream. + - name: deployment-name options: - name: name