Added server-side streams for workflows and external clients.

Makefile

@@ -522,14 +522,18 @@ workflowcheck: $(WORKFLOWCHECK)
 check: lint shell-check
 
 ##### Tests #####
+# verify-test-log reads the whole file, so a log left behind by an earlier run
+# would be judged as if it belonged to this one.
 clean-test-output:
 	@printf $(COLOR) "Delete test output..."
 	@rm -rf $(TEST_OUTPUT_ROOT)
+	@rm -f test.log
 	@go clean -testcache
 
 build-tests:
 	@printf $(COLOR) "Build tests..."
-	@CGO_ENABLED=$(CGO_ENABLED) go test $(TEST_TAG_FLAG) -exec="true" -count=0 $(TEST_DIRS)
+	@CGO_ENABLED=$(CGO_ENABLED) go test $(TEST_TAG_FLAG) -exec="true" -count=0 $(filter-out $(MIXED_BRAIN_TEST_ROOT)%,$(TEST_DIRS))
+	@cd $(MIXED_BRAIN_TEST_ROOT) && CGO_ENABLED=1 go test $(TEST_TAG_FLAG) -exec="true" -count=0 ./...
 
 unit-test: clean-test-output
 	@printf $(COLOR) "Run unit tests..."

chasm/lib/stream/config.go

@@ -0,0 +1,69 @@
+package stream
+
+import "time"
+
+// DefaultMaxMessagesPerPoll bounds a read page when the caller does not.
+const DefaultMaxMessagesPerPoll = 1000
+
+// MaxMessagesPerBatch bounds one append. It is not only an admission limit: a
+// node ID is the first offset of its batch, so to serve a read starting inside
+// a batch the reader has to find the node that contains it. Bounding the batch
+// bounds how far back it has to start, which turns an unbounded scan into a
+// fixed overread.
+const MaxMessagesPerBatch = 1000
+
+// LongPollTimeout matches the convention used by the history long polls: on
+// expiry the caller gets an empty response and polls again, rather than an
+// error it would have to special-case.
+const LongPollTimeout = 20 * time.Second
+
+// LongPollBuffer leaves room to return an empty response before the caller's
+// own deadline fires.
+const LongPollBuffer = 3 * time.Second
+
+// MaxConsumeItemsPerTask bounds one Workflow Task's slice. A byte cap alone is
+// not enough: a burst of tiny messages stays under it while still making one
+// task's drain arbitrarily long. Whichever bound binds first, the rest is
+// delivered on the following task.
+const MaxConsumeItemsPerTask = 1000
+
+// MaxConsumeBytesPerTask bounds one Workflow Task's slice by size. Paired with
+// MaxConsumeItemsPerTask because neither bound alone is enough: a burst of tiny
+// messages slips under the byte budget, and a few large ones slip under the
+// item count.
+const MaxConsumeBytesPerTask = 2 << 20
+
+// MaxProducersPerStream bounds the per-producer dedup table. The table is part
+// of the component state written on every append, so a caller that sends a
+// fresh producer id per request would grow the state until the mutable-state
+// size limit rejects every further append, leaving the stream unwritable for
+// good. The bound turns that into a clear error on the offending call.
+const MaxProducersPerStream = 1000
+
+// MaxConsumersPerStream bounds the registered consumer table for the same
+// reason. Each consumer also holds a truncation floor, so an unbounded table
+// would pin storage as well as grow state.
+const MaxConsumersPerStream = 1000
+
+// MaxListPageSize bounds a visibility page when the caller does not.
+const MaxListPageSize = 1000
+
+// MaxMessageBytes bounds one message. A message is never split, so this is also
+// the smallest unit a reader can be asked to materialise.
+const MaxMessageBytes = 1 << 20
+
+// MaxBatchBytes bounds one append. It is deliberately equal to
+// MaxConsumeBytesPerTask: a batch is written as one node and read back whole,
+// so a batch larger than a task's byte budget could never be delivered.
+const MaxBatchBytes = MaxConsumeBytesPerTask
+
+// MaxOwnedStreamsPerWorkflow bounds how many named streams one execution can
+// carry. Each is a component in the workflow's mutable state, so an unbounded
+// count grows that state until the size limit terminates the execution. The
+// name comes from the caller, and any caller in the namespace can pick a new
+// one, which is what makes this reachable from outside.
+const MaxOwnedStreamsPerWorkflow = 100
+
+// MaxStreamNameLength bounds a name before it becomes a map key in mutable
+// state, for the same reason.
+const MaxStreamNameLength = 255

chasm/lib/stream/cursor.go

@@ -0,0 +1,167 @@
+package stream
+
+import (
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/chasm"
+	streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1"
+)
+
+// Cursor is a consuming workflow's position in a stream.
+//
+// It is a subcomponent of the consumer, not of the stream. That placement is
+// the whole point: the consumer's mutable state and its History events commit
+// in one transaction, so folding a delivered range into the cursor lands
+// atomically with the event that records the range. Holding the cursor on the
+// stream instead would make every advance a cross-execution write, and a crash
+// between the two writes would either redeliver a range or skip it silently.
+type Cursor struct {
+	chasm.UnimplementedComponent
+
+	State *streampb.WorkflowStreamCursor
+}
+
+type NewCursorRequest struct {
+	StreamID string
+	// External marks a stream in another execution, whose frontier this
+	// workflow is told about rather than reads.
+	External     bool
+	CollectionID string
+	BucketSize   int64
+
+	// Where to start reading. Resolving "from the tail" against the stream's
+	// head happens before this is called, so the value recorded here is already
+	// a fact rather than a reading that would differ on replay.
+	StartOffset int64
+}
+
+func NewCursor(_ chasm.MutableContext, req NewCursorRequest) (*Cursor, error) {
+	if req.StreamID == "" {
+		return nil, serviceerror.NewInvalidArgument("stream id is required")
+	}
+	if req.CollectionID == "" {
+		return nil, serviceerror.NewInvalidArgument("collection id is required")
+	}
+	if req.BucketSize <= 0 {
+		return nil, serviceerror.NewInvalidArgument("bucket size must be positive")
+	}
+	if req.StartOffset < 0 {
+		return nil, serviceerror.NewInvalidArgument("start offset cannot be negative")
+	}
+
+	return &Cursor{
+		State: &streampb.WorkflowStreamCursor{
+			StreamId:     req.StreamID,
+			CollectionId: req.CollectionID,
+			BucketSize:   req.BucketSize,
+			Offset:       req.StartOffset,
+			StartOffset:  req.StartOffset,
+			External:     req.External,
+			KnownHead:    req.StartOffset,
+		},
+	}, nil
+}
+
+// LifecycleState reports the cursor as running for as long as the workflow
+// holding it exists. Deregistration is an explicit act, not a state the
+// component reaches on its own.
+func (c *Cursor) LifecycleState(_ chasm.Context) chasm.LifecycleState {
+	return chasm.LifecycleStateRunning
+}
+
+// Offset is the next offset that has not yet been delivered and folded in.
+func (c *Cursor) Offset() int64 {
+	return c.State.Offset
+}
+
+func (c *Cursor) StreamID() string {
+	return c.State.StreamId
+}
+
+func (c *Cursor) CollectionID() string {
+	return c.State.CollectionId
+}
+
+func (c *Cursor) BucketSize() int64 {
+	return c.State.BucketSize
+}
+
+// StagePending records the range attached to the workflow task now in flight.
+//
+// A redelivery overwrites whatever was staged before. That is safe because a
+// range only becomes history when the task completes: if the previous task
+// failed or timed out, nothing was recorded, so the replacement range is the
+// first one the workflow will ever have observed at this point.
+func (c *Cursor) StagePending(_ chasm.MutableContext, from int64, to int64) error {
+	if from < c.State.Offset {
+		return serviceerror.NewInvalidArgumentf(
+			"cannot deliver from offset %d, cursor is already at %d", from, c.State.Offset)
+	}
+	if to < from {
+		return serviceerror.NewInvalidArgumentf("range end %d precedes range start %d", to, from)
+	}
+
+	c.State.PendingFrom = from
+	c.State.PendingTo = to
+	c.State.HasPending = true
+	return nil
+}
+
+// Pending reports the staged range. The second result distinguishes "no task in
+// flight" from "a task in flight that was given nothing", which are different
+// facts: the latter must still be recorded.
+func (c *Cursor) Pending() (from int64, to int64, ok bool) {
+	if !c.State.HasPending {
+		return 0, 0, false
+	}
+	return c.State.PendingFrom, c.State.PendingTo, true
+}
+
+// Commit folds the staged range into the cursor and returns it for recording.
+// Caller writes the returned range onto the event that closes the task, in the
+// same transaction that persists this advance.
+func (c *Cursor) Commit(_ chasm.MutableContext) (from int64, to int64, ok bool) {
+	if !c.State.HasPending {
+		return 0, 0, false
+	}
+
+	from, to = c.State.PendingFrom, c.State.PendingTo
+	c.State.Offset = to
+	c.State.PendingFrom = 0
+	c.State.PendingTo = 0
+	c.State.HasPending = false
+	return from, to, true
+}
+
+// Abandon drops a staged range without advancing, for a task that will never
+// complete. The next delivery re-reads from the unchanged cursor.
+func (c *Cursor) Abandon(_ chasm.MutableContext) {
+	c.State.PendingFrom = 0
+	c.State.PendingTo = 0
+	c.State.HasPending = false
+}
+
+// IsExternal reports whether the stream lives in another execution.
+func (c *Cursor) IsExternal() bool {
+	return c.State.External
+}
+
+// KnownHead is the stream's frontier as last pushed to this workflow.
+func (c *Cursor) KnownHead() int64 {
+	return c.State.KnownHead
+}
+
+// AdvanceKnownHead moves the recorded frontier forward. It never moves back: a
+// stale push arriving after a fresher one must not hide offsets already known
+// to exist.
+func (c *Cursor) AdvanceKnownHead(_ chasm.MutableContext, head int64) {
+	if head > c.State.KnownHead {
+		c.State.KnownHead = head
+	}
+}
+
+// StartOffset is where this subscription began reading. Replay needs it to tell
+// a consumer that has committed nothing apart from one whose recording events
+// are simply not in the history page it was handed.
+func (c *Cursor) StartOffset() int64 {
+	return c.State.StartOffset
+}

chasm/lib/stream/cursor_test.go

@@ -0,0 +1,118 @@
+package stream
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1"
+)
+
+func newTestCursor(offset int64) *Cursor {
+	return &Cursor{
+		State: &streampb.WorkflowStreamCursor{
+			StreamId:     "s-1",
+			CollectionId: "col-1",
+			BucketSize:   DefaultBucketSize,
+			Offset:       offset,
+		},
+	}
+}
+
+func TestNewCursorRejectsIncompleteRequests(t *testing.T) {
+	cases := map[string]NewCursorRequest{
+		"no stream id":     {CollectionID: "col-1", BucketSize: 10},
+		"no collection id": {StreamID: "s-1", BucketSize: 10},
+		"zero bucket size": {StreamID: "s-1", CollectionID: "col-1"},
+		"negative start":   {StreamID: "s-1", CollectionID: "col-1", BucketSize: 10, StartOffset: -1},
+	}
+
+	for name, req := range cases {
+		t.Run(name, func(t *testing.T) {
+			_, err := NewCursor(nil, req)
+			require.Error(t, err)
+		})
+	}
+}
+
+func TestCursorCommitAdvancesAndClears(t *testing.T) {
+	c := newTestCursor(4)
+
+	require.NoError(t, c.StagePending(nil, 4, 7))
+
+	from, to, ok := c.Pending()
+	require.True(t, ok)
+	require.Equal(t, int64(4), from)
+	require.Equal(t, int64(7), to)
+
+	from, to, ok = c.Commit(nil)
+	require.True(t, ok)
+	require.Equal(t, int64(4), from)
+	require.Equal(t, int64(7), to)
+	require.Equal(t, int64(7), c.Offset())
+
+	_, _, ok = c.Pending()
+	require.False(t, ok, "a committed range must not be staged twice")
+
+	_, _, ok = c.Commit(nil)
+	require.False(t, ok, "committing again must not re-record the range")
+}
+
+// The distinction this pins is the one §8.2 of the design turns on: a task that
+// observed nothing still has to be recorded, so an empty range is a pending
+// range, not the absence of one.
+func TestCursorTreatsAnEmptyRangeAsAFact(t *testing.T) {
+	c := newTestCursor(9)
+
+	_, _, ok := c.Pending()
+	require.False(t, ok, "no task in flight yet")
+
+	require.NoError(t, c.StagePending(nil, 9, 9))
+
+	from, to, ok := c.Pending()
+	require.True(t, ok, "a task given nothing is still a task that must be recorded")
+	require.Equal(t, from, to)
+
+	from, to, ok = c.Commit(nil)
+	require.True(t, ok)
+	require.Equal(t, int64(9), from)
+	require.Equal(t, int64(9), to)
+	require.Equal(t, int64(9), c.Offset(), "an empty range must not move the cursor")
+}
+
+func TestCursorRejectsARangeBehindItself(t *testing.T) {
+	c := newTestCursor(12)
+
+	err := c.StagePending(nil, 11, 14)
+	require.ErrorContains(t, err, "cursor is already at 12")
+
+	err = c.StagePending(nil, 12, 11)
+	require.ErrorContains(t, err, "precedes range start")
+}
+
+// A task that failed recorded nothing, so the range it was given never became
+// history and the replacement is free to differ.
+func TestCursorRedeliveryReplacesTheStagedRange(t *testing.T) {
+	c := newTestCursor(2)
+
+	require.NoError(t, c.StagePending(nil, 2, 5))
+	require.NoError(t, c.StagePending(nil, 2, 9))
+
+	from, to, ok := c.Pending()
+	require.True(t, ok)
+	require.Equal(t, int64(2), from)
+	require.Equal(t, int64(9), to)
+	require.Equal(t, int64(2), c.Offset(), "staging alone must never advance the cursor")
+}
+
+func TestCursorAbandonLeavesTheOffsetAlone(t *testing.T) {
+	c := newTestCursor(3)
+
+	require.NoError(t, c.StagePending(nil, 3, 8))
+	c.Abandon(nil)
+
+	_, _, ok := c.Pending()
+	require.False(t, ok)
+	require.Equal(t, int64(3), c.Offset())
+
+	require.NoError(t, c.StagePending(nil, 3, 6), "the next delivery re-reads from the unchanged cursor")
+}