Let one record stream carry independent callers safely

README.md

@@ -4,7 +4,7 @@ Shared public schemas, generated Go types and validation for Weir servers and
 clients. Requires Go 1.27.1.
 
 ```sh
-go get github.com/batchstream/weir-protocol@v0.15.0
+go get github.com/batchstream/weir-protocol@v0.16.0
 ```
 
 The dependency direction is one-way: Weir and the Go SDK depend on this module.
@@ -35,10 +35,44 @@ adapter-owned formats; new adapters need no protobuf fields or schema changes.
 Empty request data is permitted. Unknown valid content types are left to the
 selected adapter.
 
-Each request is validated before execution. Earlier requests may have effects
-when a later request fails validation. The first record index is 1; every later
-request advances it by one. UINT64_MAX is invalid. Every record produces one
-ExecuteResponse with its ordinal and typed result, in input order.
+Each request is validated before execution. A record the server cannot admit,
+such as one its adapter rejects or one sent while the Store drains, receives its
+own failed result, NOT_STARTED for a mutation, and the stream continues. A
+request that fails `ValidateExecuteRequest`, changes the Store or kind, breaks
+the index sequence or cancels a record not yet sent fails the RPC with
+INVALID_ARGUMENT after the results of the records admitted before it. A record
+without a result when the RPC fails may still have effects. The first record
+index is 1; every later record advances it by one. UINT64_MAX is invalid. Each
+record receives at most one ExecuteResponse with its ordinal and typed result,
+in completion order, so consumers match results by index. One stream can
+therefore carry the records of independent callers without a slow or failed
+record holding back the others. A client that shares a stream validates each
+record before sending it, since an invalid request ends the stream for every
+caller on it.
+
+`timeout_micros` optionally bounds one Read or Mutate record from the moment
+the server reads it; time spent in transport buffers is not counted, so clients
+still enforce their own deadline. A record still queued when it expires fails
+with DEADLINE_EXCEEDED, NOT_STARTED for a mutation; an executing record reports
+the evidence its cancellation leaves. Zero sets no record deadline. Native
+requires zero and uses the RPC deadline. `RecordTimeout` and `TimeoutMicros`
+convert between the field and Go durations without shortening or disabling a
+caller's budget.
+
+A request that sets only `store_name` and `cancel_index` cancels an earlier
+record of the same stream. It does not advance the index or receive a response.
+The canceled record still receives its own result: a CANCELLED failure,
+NOT_STARTED for a mutation, while it is still queued; the evidence its
+cancellation leaves while it executes; and its ordinary result once it has
+completed. Cancel requests let a stream shared by several callers release the
+work of one caller who stops waiting. Native streams use RPC cancellation
+instead.
+
+Each stream has room for `RecordWindow` (32) records without results, whatever
+the Store queue holds. A client that never has more than 32 records without
+results is never held back by the server's window, so its cancel requests take
+effect as soon as they arrive; a further record waits until a result frees room.
+
 Servers and clients enforce Store/kind consistency and index sequences. Stateless
 validators check each envelope and nested fields. Requests and responses reject unknown
 fields throughout the message tree. Responses require declared result/event

api/protocol/protocol.go

@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"regexp"
 	"strings"
+	"time"
 	"unicode"
 	"unicode/utf8"
 
@@ -20,6 +21,11 @@ import (
 // Backend protocol limits and actual process capacity govern business payloads.
 const MaxTransportMessageBytes = int(^uint(0) >> 1)
 
+// RecordWindow is the number of records without results each stream has room
+// for. A client that never exceeds it is never held back by the server's
+// window, so its cancel requests take effect once they arrive.
+const RecordWindow = 32
+
 var storePattern = regexp.MustCompile(`^[a-z](?:[a-z0-9]|-[a-z0-9])*$`)
 var mediaPattern = regexp.MustCompile(`^[a-z0-9!#$&^_.+-]+/[a-z0-9!#$&^_.+-]+$`)
 
@@ -197,9 +203,13 @@ func NativeFailure(started bool, failure *pb.Failure) *pb.NativeEnd {
 	return end
 }
 
-// ValidateExecuteRequest validates one operation without retaining stream
-// state. Callers enforce a fixed Store/kind and consecutive indexes across requests.
+// ValidateExecuteRequest validates one record or cancel request without
+// retaining stream state. Callers enforce a fixed Store/kind, consecutive
+// indexes and cancel requests that name earlier records.
 func ValidateExecuteRequest(req *pb.ExecuteRequest) error {
+	if req != nil && req.CancelIndex != 0 {
+		return validateCancelRequest(req)
+	}
 	if req == nil || !ValidStoreName(req.StoreName) || req.Index == 0 {
 		return fmt.Errorf("invalid Execute request envelope")
 	}
@@ -215,16 +225,50 @@ func ValidateExecuteRequest(req *pb.ExecuteRequest) error {
 			return fmt.Errorf("record index overflows the next ordinal")
 		}
 	default:
-		if req.Index != 1 {
-			return fmt.Errorf("Native requires index 1")
+		if req.Index != 1 || req.TimeoutMicros != 0 {
+			return fmt.Errorf("Native requires index 1 and no record timeout")
 		}
 	}
 	return nil
 }
 
+func validateCancelRequest(req *pb.ExecuteRequest) error {
+	if !ValidStoreName(req.StoreName) || req.CancelIndex == math.MaxUint64 || req.Index != 0 || req.Command != nil || req.TimeoutMicros != 0 {
+		return fmt.Errorf("a cancel request carries only its Store and cancel_index")
+	}
+	if hasUnknown(req.ProtoReflect()) {
+		return fmt.Errorf("invalid Execute request fields")
+	}
+	return nil
+}
+
+// RecordTimeout converts a request's timeout_micros, saturating at the
+// longest representable duration. Zero means no record deadline.
+func RecordTimeout(micros uint64) time.Duration {
+	if micros > uint64(math.MaxInt64/int64(time.Microsecond)) {
+		return math.MaxInt64
+	}
+	return time.Duration(micros) * time.Microsecond
+}
+
+// TimeoutMicros rounds a remaining caller budget up to whole microseconds so a
+// record never receives less time than its caller allows. An exhausted budget
+// maps to one microsecond, never to the zero that disables the deadline.
+func TimeoutMicros(remaining time.Duration) uint64 {
+	if remaining <= time.Microsecond {
+		return 1
+	}
+	micros := uint64(remaining / time.Microsecond)
+	if remaining%time.Microsecond != 0 {
+		micros++
+	}
+	return micros
+}
+
 // ValidateExecuteResponse validates one indexed event and rejects unknown fields
-// recursively. Stream consumers enforce the expected event kind, result index sequence,
-// and terminal RPC evidence. Missing or unrecognized event variants remain invalid.
+// recursively. Stream consumers enforce the expected event kind, at most one
+// result per submitted index, and terminal RPC evidence. Missing or
+// unrecognized event variants remain invalid.
 func ValidateExecuteResponse(response *pb.ExecuteResponse) error {
 	if response == nil || response.Index == 0 || response.Index == math.MaxUint64 || hasUnknown(response.ProtoReflect()) {
 		return fmt.Errorf("invalid Execute response envelope")
@@ -320,22 +364,29 @@ func hasUnknown(message protoreflect.Message) bool {
 	if len(message.GetUnknown()) != 0 {
 		return true
 	}
-	unknown := false
-	message.Range(func(field protoreflect.FieldDescriptor, value protoreflect.Value) bool {
-		if field.IsList() && field.Message() != nil {
-			list := value.List()
-			for i := 0; i < list.Len(); i++ {
-				if hasUnknown(list.Get(i).Message()) {
-					unknown = true
-					return false
-				}
+	// Walk populated message fields by descriptor; a Range callback would
+	// allocate on every validated message.
+	fields := message.Descriptor().Fields()
+	for i := range fields.Len() {
+		field := fields.Get(i)
+		if field.Message() == nil || !message.Has(field) {
+			continue
+		}
+		value := message.Get(field)
+		if !field.IsList() {
+			if hasUnknown(value.Message()) {
+				return true
 			}
-		} else if field.Message() != nil {
-			unknown = hasUnknown(value.Message())
+			continue
 		}
-		return !unknown
-	})
-	return unknown
+		list := value.List()
+		for j := range list.Len() {
+			if hasUnknown(list.Get(j).Message()) {
+				return true
+			}
+		}
+	}
+	return false
 }
 
 // ValidateEvent checks known event semantics and fields.

api/protocol/protocol_test.go

@@ -1,8 +1,10 @@
 package protocol
 
 import (
+	"math"
 	"strings"
 	"testing"
+	"time"
 	"unicode/utf8"
 
 	pb "github.com/batchstream/weir-protocol/api/weir/v1"
@@ -113,6 +115,46 @@ func TestExecuteTypedEnvelopeAndUnknownFields(t *testing.T) {
 	}
 }
 
+func TestCancelRequestCarriesOnlyItsStoreAndIndex(t *testing.T) {
+	request := &pb.ExecuteRequest{StoreName: "records", CancelIndex: 3}
+	if err := ValidateExecuteRequest(request); err != nil {
+		t.Fatal("cancel request rejected", err)
+	}
+	invalid := []func(*pb.ExecuteRequest){
+		func(r *pb.ExecuteRequest) { r.StoreName = "" },
+		func(r *pb.ExecuteRequest) { r.CancelIndex = math.MaxUint64 },
+		func(r *pb.ExecuteRequest) { r.ProtoReflect().SetUnknown([]byte{0xf8, 0x07, 1}) },
+	}
+	// Every other field, including one added later, must stay unset.
+	fields := request.ProtoReflect().Descriptor().Fields()
+	for i := range fields.Len() {
+		field := fields.Get(i)
+		if field.Name() == "store_name" || field.Name() == "cancel_index" {
+			continue
+		}
+		invalid = append(invalid, func(r *pb.ExecuteRequest) {
+			message := r.ProtoReflect()
+			switch field.Kind() {
+			case protoreflect.MessageKind:
+				message.Set(field, protoreflect.ValueOfMessage(message.NewField(field).Message()))
+			case protoreflect.StringKind:
+				message.Set(field, protoreflect.ValueOfString("x"))
+			case protoreflect.Uint64Kind:
+				message.Set(field, protoreflect.ValueOfUint64(1))
+			default:
+				t.Fatal("cancel test cannot populate field", field.FullName())
+			}
+		})
+	}
+	for _, mutate := range invalid {
+		copied := proto.Clone(request).(*pb.ExecuteRequest)
+		mutate(copied)
+		if err := ValidateExecuteRequest(copied); err == nil {
+			t.Fatal("accepted invalid cancel request", copied)
+		}
+	}
+}
+
 func TestNativeUsesOneIndexedFrame(t *testing.T) {
 	body := &pb.Document{ContentType: "application/vnd.example.native", Data: []byte("command")}
 	native := &pb.NativeRequest{Resource: "records", Request: body}
@@ -126,6 +168,10 @@ func TestNativeUsesOneIndexedFrame(t *testing.T) {
 	if err := ValidateExecuteRequest(request); err == nil {
 		t.Fatal("Native record ordinal accepted")
 	}
+	request.Index, request.TimeoutMicros = 1, 1
+	if err := ValidateExecuteRequest(request); err == nil {
+		t.Fatal("Native record timeout accepted")
+	}
 	head := &pb.NativeHead{}
 	value := &pb.Event_Head{Head: head}
 	event := &pb.Event{Value: value}
@@ -212,3 +258,43 @@ func TestSchemaHasNoMapFields(t *testing.T) {
 		}
 	}
 }
+
+func TestRecordTimeoutNeverShortensOrDisablesTheCallerBudget(t *testing.T) {
+	read := &pb.ReadRequest{Resource: "records/a"}
+	request := readExecution("records", 1, read)
+	request.TimeoutMicros = math.MaxUint64
+	if err := ValidateExecuteRequest(request); err != nil {
+		t.Fatal("record timeout rejected", err)
+	}
+	cases := []struct {
+		remaining time.Duration
+		micros    uint64
+	}{
+		{remaining: -time.Second, micros: 1},
+		{remaining: 0, micros: 1},
+		{remaining: time.Nanosecond, micros: 1},
+		{remaining: time.Microsecond + time.Nanosecond, micros: 2},
+		{remaining: 3 * time.Second, micros: 3_000_000},
+		{remaining: math.MaxInt64, micros: math.MaxInt64/1000 + 1},
+	}
+	for _, c := range cases {
+		if got := TimeoutMicros(c.remaining); got != c.micros {
+			t.Fatalf("TimeoutMicros(%v) = %d, want %d", c.remaining, got, c.micros)
+		}
+	}
+	durations := []struct {
+		micros    uint64
+		remaining time.Duration
+	}{
+		{micros: 0, remaining: 0},
+		{micros: 3_000_000, remaining: 3 * time.Second},
+		{micros: math.MaxInt64 / 1000, remaining: math.MaxInt64 / 1000 * time.Microsecond},
+		{micros: math.MaxInt64/1000 + 1, remaining: math.MaxInt64},
+		{micros: math.MaxUint64, remaining: math.MaxInt64},
+	}
+	for _, c := range durations {
+		if got := RecordTimeout(c.micros); got != c.remaining {
+			t.Fatalf("RecordTimeout(%d) = %v, want %v", c.micros, got, c.remaining)
+		}
+	}
+}

api/weir/v1/contract_test.go

@@ -61,7 +61,7 @@ func TestPublicStoreContract(t *testing.T) {
 	if StoreService_Execute_FullMethodName != "/weir.v1.StoreService/Execute" {
 		t.Fatal("generated public method path changed")
 	}
-	executeRequest := []fieldContract{{name: "store_name", kind: protoreflect.StringKind}, {name: "index", kind: protoreflect.Uint64Kind}, {name: "command", kind: protoreflect.MessageKind, message: "weir.v1.Command"}}
+	executeRequest := []fieldContract{{name: "store_name", kind: protoreflect.StringKind}, {name: "index", kind: protoreflect.Uint64Kind}, {name: "command", kind: protoreflect.MessageKind, message: "weir.v1.Command"}, {name: "timeout_micros", kind: protoreflect.Uint64Kind}, {name: "cancel_index", kind: protoreflect.Uint64Kind}}
 	assertFields(t, execute.Input(), executeRequest)
 	executeResponse := []fieldContract{{name: "index", kind: protoreflect.Uint64Kind}, {name: "event", kind: protoreflect.MessageKind, message: "weir.v1.Event"}}
 	assertFields(t, execute.Output(), executeResponse)