From f9bcbe8c29dac7e8fc04969ae5c773ebd7308362 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 03:35:33 -0700 Subject: [PATCH 1/5] Kept the connection the SDK dials for services it does not wrap. The stream service has no SDK client, so its commands need the dialed connection with its TLS, credentials, headers and interceptors. Dialing makes a call, so an interceptor captures it. --- internal/temporalcli/client.go | 52 ++++++++++++++++++++++++++++++---- 1 file changed, 47 insertions(+), 5 deletions(-) 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 From c8073be9ebc00859e9b92f8af3030e5f77595a99 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 03:36:06 -0700 Subject: [PATCH 2/5] Added the temporal stream command group. Create, list, describe, read, append, close, truncate and delete reach a standalone Stream or one a Workflow or Activity owns. Describe also names the channel the Stream notifies. --- internal/temporalcli/commands.gen.go | 322 +++++++ internal/temporalcli/commands.stream.go | 1020 +++++++++++++++++++++++ internal/temporalcli/commands.yaml | 353 ++++++++ 3 files changed, 1695 insertions(+) create mode 100644 internal/temporalcli/commands.stream.go diff --git a/internal/temporalcli/commands.gen.go b/internal/temporalcli/commands.gen.go index ed56d7cc1..df9e2bf23 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) @@ -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.yaml b/internal/temporalcli/commands.yaml index 18b1020fe..935320a9a 100644 --- a/internal/temporalcli/commands.yaml +++ b/internal/temporalcli/commands.yaml @@ -3710,6 +3710,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 +5978,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 From 296bb5acadcf4a69cab3fdc596018c6d81dd57d1 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 03:36:19 -0700 Subject: [PATCH 3/5] Turned streams on in the dev server. The server ships streams off so a deployment opts in, and the dev server exists for local development. A dynamic config value still turns them off. --- internal/devserver/server.go | 5 +++++ internal/temporalcli/commands.gen.go | 4 ++-- internal/temporalcli/commands.yaml | 8 ++++++++ 3 files changed, 15 insertions(+), 2 deletions(-) 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/commands.gen.go b/internal/temporalcli/commands.gen.go index df9e2bf23..59cc77623 100644 --- a/internal/temporalcli/commands.gen.go +++ b/internal/temporalcli/commands.gen.go @@ -3103,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.") diff --git a/internal/temporalcli/commands.yaml b/internal/temporalcli/commands.yaml index 935320a9a..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: From 4bac9eb42ecd477441ee1fb5504eb0cfd248e68f Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 03:36:25 -0700 Subject: [PATCH 4/5] Documented the temporal stream commands. The page covers addressing, the refusals and the dev server switch in one place for people trying streams from a terminal. --- docs/stream.md | 145 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 145 insertions(+) create mode 100644 docs/stream.md 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 +``` From 3c9bd3417502023a4fab4aa6a4db0ee255a396ac Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 3 Oct 2026 03:36:56 -0700 Subject: [PATCH 5/5] Tested the temporal stream commands. Fake-service cases pin the channel card for each owner kind, and the shared dev server cases run the stream lifecycle, an owned Stream and the channel a Stream notifies. --- internal/temporalcli/commands.channel_test.go | 36 +- internal/temporalcli/commands.stream_test.go | 477 ++++++++++++++++++ 2 files changed, 512 insertions(+), 1 deletion(-) create mode 100644 internal/temporalcli/commands.stream_test.go 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.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") +}