Notified the channel named by a native stream on each change.

chasm/lib/activity/streams.go

@@ -40,5 +40,14 @@ func (a *Activity) AppendToOwnedStream(
 			"activity execution closed with status %v, so its streams take no more records",
 			InternalStatusToAPIStatus(a.GetStatus()))
 	}
-	return a.ownedStreams().Append(mctx, key, req)
+	result, err := a.ownedStreams().Append(mctx, key, req)
+	if err != nil || result.Deduplicated {
+		return result, err
+	}
+	// A standalone activity has no linked channels, so its streams notify an
+	// independent channel named by the activity and the stream.
+	owner := mctx.ExecutionKey()
+	a.ownedStreams().Get(mctx, key).ScheduleChannelNotify(mctx,
+		stream.ActivityChannelName(owner.BusinessID, key), owner.RunID, req.Limits)
+	return result, nil
 }

chasm/lib/channel/linked.go

@@ -54,6 +54,27 @@ func (c *Channel) NotifyLinked(
 	mctx chasm.MutableContext,
 	n *channelpb.Notification,
 	limits Limits,
+) (LinkedNotifyResult, error) {
+	return c.notifyLinked(mctx, n, limits, true)
+}
+
+// NotifyLinkedListeners accepts a notification for the ring and the callback
+// listeners without handing it to the owner. The channel a stream drives
+// uses it: the owning run learns of its own stream through its cursor, and a
+// Workflow Task per append would be a cost the stream did not have.
+func (c *Channel) NotifyLinkedListeners(
+	mctx chasm.MutableContext,
+	n *channelpb.Notification,
+	limits Limits,
+) (LinkedNotifyResult, error) {
+	return c.notifyLinked(mctx, n, limits, false)
+}
+
+func (c *Channel) notifyLinked(
+	mctx chasm.MutableContext,
+	n *channelpb.Notification,
+	limits Limits,
+	wakeOwner bool,
 ) (LinkedNotifyResult, error) {
 	limits = limits.withDefaults()
 	if err := CheckNotification(n, limits.MaxMetadataBytes); err != nil {
@@ -69,9 +90,12 @@ func (c *Channel) NotifyLinked(
 		out.Advanced = true
 	}
 	c.touch(mctx)
-	if c.OwnerHolds(n.GetCounter()) {
+	switch {
+	case !wakeOwner:
+		out.OwnerHeld = true
+	case c.OwnerHolds(n.GetCounter()):
 		out.OwnerHeld = true
-	} else {
+	default:
 		out.OwnerFolded = c.State.GetOwnerPending() != nil
 		c.State.OwnerPending = n
 	}

chasm/lib/channel/linked_test.go

@@ -46,6 +46,28 @@ func TestLinkedNotifyReachesTheOwnerInOneWrite(t *testing.T) {
 	require.Equal(t, "owner", polled[0].GetLinkedTo().GetWorkflowId())
 }
 
+// A notify for the listeners alone joins the ring and reaches the callbacks,
+// and leaves the owner with nothing pending: the channel a stream drives
+// must not wake the run for its own stream.
+func TestLinkedNotifyListenersLeavesTheOwnerAlone(t *testing.T) {
+	c, mctx := newTestLinkedChannel(t)
+
+	result, err := c.NotifyLinkedListeners(mctx, note(1), Limits{})
+	require.NoError(t, err)
+	require.True(t, result.Advanced)
+	require.True(t, result.OwnerHeld, "nothing for the owner to take")
+	require.False(t, c.HasOwnerPending())
+	require.Equal(t, int64(1), c.LatestCounter())
+	polled, err := c.Poll(mctx, PollRequest{})
+	require.NoError(t, err)
+	require.Equal(t, []int64{1}, counters(polled))
+	require.Equal(t, "owner", polled[0].GetLinkedTo().GetWorkflowId())
+
+	_, err = c.NotifyLinked(mctx, note(2), Limits{})
+	require.NoError(t, err)
+	require.Equal(t, int64(2), c.State.GetOwnerPending().GetCounter(), "a plain notify still wakes")
+}
+
 // The owner folds like a subscribed listener: a repeat at or below what it
 // holds pending, or what its unstarted task carries, changes nothing; a
 // higher counter replaces the pending entry; and once the task has started

chasm/lib/stream/channel.go

@@ -0,0 +1,105 @@
+package stream
+
+import (
+	"strconv"
+
+	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/server/chasm"
+	channelpb "go.temporal.io/server/chasm/lib/channel/gen/channelpb/v1"
+	streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1"
+	"go.temporal.io/server/common/payload"
+)
+
+// A change to a stream notifies the notification channel named by the stream,
+// so callback and polling listeners learn of a native stream the way they
+// learn of an external one. The name is derived from the stream's identity
+// alone, so a client derives the same name without asking the server. The
+// workflow consumers of a stream keep their cursors and are not told twice.
+
+// ChannelNamePrefix starts the name of every channel a stream drives.
+const ChannelNamePrefix = "stream/"
+
+// ChannelName is the independent channel a standalone stream notifies.
+func ChannelName(streamID string) string {
+	return ChannelNamePrefix + streamID
+}
+
+// OwnedChannelName is the channel, linked to the owning run, that a stream a
+// workflow owns notifies.
+func OwnedChannelName(name string) string {
+	return ChannelNamePrefix + name
+}
+
+// ActivityChannelName is the channel a stream an activity owns notifies: linked
+// to the workflow run for an activity a workflow scheduled, independent for a
+// standalone activity, which has no linked channels of its own.
+func ActivityChannelName(activityID, name string) string {
+	return ChannelNamePrefix + activityID + "/" + name
+}
+
+// ClosedMetadataKey is set, to true, on the notification for a close.
+const ClosedMetadataKey = "closed"
+
+// ClosedMetadata is the metadata of a close notification.
+func ClosedMetadata() map[string]*commonpb.Payload {
+	// A bool encodes without error.
+	closed, _ := payload.Encode(true)
+	return map[string]*commonpb.Payload{ClosedMetadataKey: closed}
+}
+
+// ChangeNotification describes the stream's latest change for its channel:
+// the change sequence is the counter, the head after the change is the
+// position in the form a native cursor names one, and a close says so.
+//
+// The run is what a native cursor pairs the offset with: the owning run of an
+// attached stream, and the stream's own id for a standalone one, which has no
+// run a client would know.
+func (s *Stream) ChangeNotification(channel, run string) *channelpb.Notification {
+	n := &channelpb.Notification{
+		Channel:  channel,
+		Counter:  s.State.GetChangeSequence(),
+		Position: EncodePosition(run, s.State.GetHeadOffset()),
+	}
+	if s.State.GetClosed() {
+		n.Metadata = ClosedMetadata()
+	}
+	return n
+}
+
+// EncodePosition is a native position as a client's cursor names it.
+func EncodePosition(run string, offset int64) []byte {
+	return []byte(run + ":" + strconv.FormatInt(offset, 10))
+}
+
+// ScheduleChannelNotify arms a task to hand the latest change to the stream's
+// channel, for a stream whose channel is an execution of its own: a
+// standalone stream's, or that of a stream a standalone activity owns. One
+// task is outstanding at a time and reads the latest change when it runs.
+// Tasks of one execution may run in any order, and a change arriving at the
+// channel below one it already holds is folded away, so a task per change
+// would lose the earlier ones from the ring; one task at a time keeps the
+// counters the channel sees in order, with a burst arriving as its newest.
+func (s *Stream) ScheduleChannelNotify(
+	mctx chasm.MutableContext,
+	channel, run string,
+	limits Limits,
+) {
+	if mctx == nil || limits.ChannelNotifyOff || s.State.GetChannelNotifyPending() {
+		return
+	}
+	s.State.ChannelNotifyPending = true
+	mctx.AddTask(s, chasm.TaskAttributes{ScheduledTime: mctx.Now(s)},
+		&streamlib.StreamNotifyChannelTask{Channel: channel, Run: run})
+}
+
+// TakeChannelChange is the channel task's read of the stream. Lowering the
+// flag in the transition that reads the change is what makes the coalescing
+// safe: a change that commits after this one sees the flag down and arms the
+// next task.
+func (s *Stream) TakeChannelChange(
+	_ chasm.MutableContext,
+	task *streamlib.StreamNotifyChannelTask,
+) (*channelpb.Notification, error) {
+	s.State.ChannelNotifyPending = false
+	return s.ChangeNotification(task.GetChannel(), task.GetRun()), nil
+}

chasm/lib/stream/channel_test.go

@@ -0,0 +1,84 @@
+package stream
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.temporal.io/server/chasm"
+	streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1"
+	"go.temporal.io/server/common/payload"
+)
+
+func newChannelTestContext() *chasm.MockMutableContext {
+	now := time.Unix(1_700_000_000, 0).UTC()
+	return &chasm.MockMutableContext{MockContext: chasm.MockContext{
+		HandleNow: func(chasm.Component) time.Time { return now },
+	}}
+}
+
+// Every append and the close are one change each, a producer's retry none,
+// and the notification for a change carries the sequence, the head paired
+// with the run, and closed only on the close.
+func TestStreamChangeNotification(t *testing.T) {
+	ctx := newChannelTestContext()
+	s := newTestStream(t)
+
+	_, err := s.AddMessages(ctx, AddMessagesRequest{Records: msgs("a", "b")})
+	require.NoError(t, err)
+	n := s.ChangeNotification("stream/s", "run-1")
+	require.Equal(t, "stream/s", n.GetChannel())
+	require.Equal(t, int64(1), n.GetCounter())
+	require.Equal(t, "run-1:2", string(n.GetPosition()))
+	require.Empty(t, n.GetMetadata())
+
+	retry := AddMessagesRequest{Records: msgs("c"), ProducerID: "p", Sequence: 1}
+	_, err = s.AddMessages(ctx, retry)
+	require.NoError(t, err)
+	replay, err := s.AddMessages(ctx, retry)
+	require.NoError(t, err)
+	require.True(t, replay.Deduplicated)
+	require.Equal(t, int64(2), s.ChangeNotification("stream/s", "run-1").GetCounter(),
+		"a replay is not a change")
+
+	require.NoError(t, s.CloseAndSchedule(ctx, nil))
+	n = s.ChangeNotification("stream/s", "run-1")
+	require.Equal(t, int64(3), n.GetCounter())
+	require.Equal(t, "run-1:3", string(n.GetPosition()))
+	var closed bool
+	require.NoError(t, payload.Decode(n.GetMetadata()[ClosedMetadataKey], &closed))
+	require.True(t, closed)
+	s.Close(ctx.Now(s), nil)
+	require.Equal(t, int64(3), s.State.GetChangeSequence(), "a second close is not a change")
+}
+
+// One channel task is outstanding at a time, it reads the latest change when
+// it runs and lowers the flag, and the switch leaves none.
+func TestStreamScheduleChannelNotify(t *testing.T) {
+	ctx := newChannelTestContext()
+	s := newTestStream(t)
+	_, err := s.AddMessages(ctx, AddMessagesRequest{Records: msgs("a")})
+	require.NoError(t, err)
+
+	s.ScheduleChannelNotify(ctx, "stream/s", "s", Limits{ChannelNotifyOff: true})
+	require.Empty(t, ctx.Tasks)
+
+	s.ScheduleChannelNotify(ctx, "stream/s", "s", Limits{})
+	require.Len(t, ctx.Tasks, 1)
+	task, ok := ctx.Tasks[0].Payload.(*streamlib.StreamNotifyChannelTask)
+	require.True(t, ok, "%T", ctx.Tasks[0].Payload)
+	require.Equal(t, "stream/s", task.GetChannel())
+	require.Equal(t, "s", task.GetRun())
+
+	_, err = s.AddMessages(ctx, AddMessagesRequest{Records: msgs("b")})
+	require.NoError(t, err)
+	s.ScheduleChannelNotify(ctx, "stream/s", "s", Limits{})
+	require.Len(t, ctx.Tasks, 1, "one task at a time")
+
+	n, err := s.TakeChannelChange(ctx, task)
+	require.NoError(t, err)
+	require.Equal(t, int64(2), n.GetCounter(), "the task reads the latest change")
+	require.Equal(t, "s:2", string(n.GetPosition()))
+	s.ScheduleChannelNotify(ctx, "stream/s", "s", Limits{})
+	require.Len(t, ctx.Tasks, 2, "the flag is down, so the next change arms a task")
+}

chasm/lib/stream/config.go

@@ -128,6 +128,11 @@ const (
 	// every single stream stays inside its own budget. Half the error limit,
 	// which leaves the rest of mutable state its own room.
 	OwnedStreamsMaxBytesPerWorkflow = 4 << 20
+
+	// ChannelRetainedNotifications bounds the ring of a channel a stream drives,
+	// well below a linked channel's default, since the stream holds the data a
+	// poller would otherwise need the ring for.
+	ChannelRetainedNotifications = 16
 )
 
 var (
@@ -141,6 +146,17 @@ var (
 		`Whether the stream service and the workflow stream commands are available to a
 namespace. Off by default.`,
 	)
+	NotifyChannelSetting = dynamicconfig.NewNamespaceBoolSetting(
+		"stream.notifyChannel",
+		true,
+		`Whether a change to a native stream notifies the notification channel named by the
+stream. Off creates no channel state from streams.`,
+	)
+	ChannelRetainedNotificationsSetting = dynamicconfig.NewNamespaceIntSetting(
+		"stream.channelRetainedNotifications",
+		ChannelRetainedNotifications,
+		`How many notifications the channel named by a native stream retains for pollers.`,
+	)
 	MaxConsumeItemsPerTaskSetting = dynamicconfig.NewNamespaceIntSetting(
 		"stream.maxConsumeItemsPerTask",
 		MaxConsumeItemsPerTask,
@@ -290,10 +306,16 @@ type Config struct {
 	AppendRecordsPerSecond dynamicconfig.IntPropertyFnWithNamespaceFilter
 	AppendBytesPerSecond   dynamicconfig.IntPropertyFnWithNamespaceFilter
 	PollsPerSecond         dynamicconfig.IntPropertyFnWithNamespaceFilter
+
+	NotifyChannel                dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	ChannelRetainedNotifications dynamicconfig.IntPropertyFnWithNamespaceFilter
 }
 
 func NewConfig(dc *dynamicconfig.Collection) *Config {
 	return &Config{
+		NotifyChannel:                NotifyChannelSetting.Get(dc),
+		ChannelRetainedNotifications: ChannelRetainedNotificationsSetting.Get(dc),
+
 		Enabled:                    EnabledSetting.Get(dc),
 		MaxIDLength:                dynamicconfig.MaxIDLengthLimit.Get(dc),
 		RetentionRecheckInterval:   RetentionRecheckIntervalSetting.Get(dc),
@@ -338,6 +360,12 @@ type Limits struct {
 	ReplayMaxRecords                int
 	ReplayMaxBytes                  int
 	ReplayMaxPages                  int
+
+	// ChannelNotifyOff stops a stream's changes from reaching the channel
+	// named by the stream. The zero value is on, so a Limits built by hand
+	// behaves as a namespace with the default setting.
+	ChannelNotifyOff             bool
+	ChannelRetainedNotifications int
 }
 
 // LimitsFor resolves the limits for a namespace. A nil Config, which is what
@@ -347,6 +375,9 @@ func (c *Config) LimitsFor(namespaceName string) Limits {
 		return DefaultLimits()
 	}
 	return Limits{
+		ChannelNotifyOff:             !c.NotifyChannel(namespaceName),
+		ChannelRetainedNotifications: c.ChannelRetainedNotifications(namespaceName),
+
 		MaxConsumeItemsPerTask:     c.MaxConsumeItemsPerTask(namespaceName),
 		MaxConsumeBytesPerTask:     c.MaxConsumeBytesPerTask(namespaceName),
 		MaxProducersPerStream:      c.MaxProducersPerStream(namespaceName),
@@ -410,5 +441,6 @@ func (l Limits) withDefaults() Limits {
 	fill(&l.ReplayMaxRecords, ReplayMaxRecords)
 	fill(&l.ReplayMaxBytes, ReplayMaxBytes)
 	fill(&l.ReplayMaxPages, ReplayMaxPages)
+	fill(&l.ChannelRetainedNotifications, ChannelRetainedNotifications)
 	return l
 }