From 48eea4ae0f0c0fe6c5a1cbf014a9770cc548f6c0 Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Fri, 30 Oct 2026 13:30:11 +0800
Subject: [PATCH 1/6] =?UTF-8?q?=E2=9A=A1=EF=B8=8F=20perf:=20check=20unknow?=
 =?UTF-8?q?n=20fields=20without=20allocating=20per=20message?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 api/protocol/protocol.go | 35 +++++++++++++++++++++--------------
 1 file changed, 21 insertions(+), 14 deletions(-)

diff --git a/api/protocol/protocol.go b/api/protocol/protocol.go
index ab55eee..cba790a 100644
--- a/api/protocol/protocol.go
+++ b/api/protocol/protocol.go
@@ -320,22 +320,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
+			}
+			continue
+		}
+		list := value.List()
+		for j := range list.Len() {
+			if hasUnknown(list.Get(j).Message()) {
+				return true
 			}
-		} else if field.Message() != nil {
-			unknown = hasUnknown(value.Message())
 		}
-		return !unknown
-	})
-	return unknown
+	}
+	return false
 }
 
 // ValidateEvent checks known event semantics and fields.

From e198f19f56c25f9942632db6b1dcf9ad3ee8f8a7 Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Fri, 30 Oct 2026 13:30:13 +0800
Subject: [PATCH 2/6] =?UTF-8?q?=E2=9C=A8=20feat:=20deliver=20record=20resu?=
 =?UTF-8?q?lts=20in=20completion=20order=20with=20per-record=20rejection?=
 =?UTF-8?q?=20and=20timeout?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 README.md                     | 22 +++++++++++++++++-----
 api/protocol/protocol.go      | 30 ++++++++++++++++++++++++++----
 api/protocol/protocol_test.go | 33 +++++++++++++++++++++++++++++++++
 api/weir/v1/contract_test.go  |  2 +-
 api/weir/v1/store.pb.go       | 22 ++++++++++++++++++----
 api/weir/v1/store.proto       | 15 ++++++++++++---
 api/weir/v1/store_grpc.pb.go  | 12 ++++++++----
 7 files changed, 115 insertions(+), 21 deletions(-)

diff --git a/README.md b/README.md
index 358db76..230cdd9 100644
--- a/README.md
+++ b/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,22 @@ 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 and the stream continues. An invalid envelope or index
+sequence ends the stream; earlier records may have effects when it does. 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 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.
+
+`timeout_micros` optionally bounds one record from the moment the server
+receives it. A record still queued when it expires is NOT_STARTED with
+DEADLINE_EXCEEDED; 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.
 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
diff --git a/api/protocol/protocol.go b/api/protocol/protocol.go
index cba790a..6368c52 100644
--- a/api/protocol/protocol.go
+++ b/api/protocol/protocol.go
@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"regexp"
 	"strings"
+	"time"
 	"unicode"
 	"unicode/utf8"
 
@@ -215,16 +216,37 @@ 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
 }
 
+// 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 := (remaining + time.Microsecond - 1) / time.Microsecond
+	return uint64(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, exactly 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")
diff --git a/api/protocol/protocol_test.go b/api/protocol/protocol_test.go
index 2120e49..213158d 100644
--- a/api/protocol/protocol_test.go
+++ b/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"
@@ -126,6 +128,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 +218,30 @@ 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},
+	}
+	for _, c := range cases {
+		if got := TimeoutMicros(c.remaining); got != c.micros {
+			t.Fatalf("TimeoutMicros(%v) = %d, want %d", c.remaining, got, c.micros)
+		}
+	}
+	if RecordTimeout(0) != 0 || RecordTimeout(3_000_000) != 3*time.Second || RecordTimeout(math.MaxUint64) != math.MaxInt64 {
+		t.Fatal("record timeout conversion changed")
+	}
+}
diff --git a/api/weir/v1/contract_test.go b/api/weir/v1/contract_test.go
index 52eaee0..fb4dff1 100644
--- a/api/weir/v1/contract_test.go
+++ b/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}}
 	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)
diff --git a/api/weir/v1/store.pb.go b/api/weir/v1/store.pb.go
index c91745b..673f076 100644
--- a/api/weir/v1/store.pb.go
+++ b/api/weir/v1/store.pb.go
@@ -296,7 +296,13 @@ type ExecuteRequest struct {
 	// it by one. Native always uses 1. UINT64_MAX is invalid.
 	Index uint64 `protobuf:"varint,2,opt,name=index,proto3" json:"index,omitempty"`
 	// Required operation; its kind must remain identical throughout the stream.
-	Command       *Command `protobuf:"bytes,3,opt,name=command,proto3" json:"command,omitempty"`
+	Command *Command `protobuf:"bytes,3,opt,name=command,proto3" json:"command,omitempty"`
+	// Optional record deadline in microseconds after the server receives the
+	// request. A record still queued when it expires is NOT_STARTED with
+	// DEADLINE_EXCEEDED; an executing record reports the evidence its
+	// cancellation leaves. Zero sets no record deadline. Native requires zero
+	// and uses the RPC deadline.
+	TimeoutMicros uint64 `protobuf:"varint,4,opt,name=timeout_micros,json=timeoutMicros,proto3" json:"timeout_micros,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -352,10 +358,17 @@ func (x *ExecuteRequest) GetCommand() *Command {
 	return nil
 }
 
+func (x *ExecuteRequest) GetTimeoutMicros() uint64 {
+	if x != nil {
+		return x.TimeoutMicros
+	}
+	return 0
+}
+
 // ExecuteResponse carries one result or streaming event.
 type ExecuteResponse struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
-	// One-based input record ordinal; record results arrive in input order.
+	// One-based input record ordinal; record results arrive in completion order.
 	// Every input record has exactly one result. Native events all use 1.
 	Index uint64 `protobuf:"varint,1,opt,name=index,proto3" json:"index,omitempty"`
 	// Required typed event; it must match the stream's operation kind.
@@ -1533,12 +1546,13 @@ var File_api_weir_v1_store_proto protoreflect.FileDescriptor
 
 const file_api_weir_v1_store_proto_rawDesc = "" +
 	"\n" +
-	"\x17api/weir/v1/store.proto\x12\aweir.v1\"q\n" +
+	"\x17api/weir/v1/store.proto\x12\aweir.v1\"\x98\x01\n" +
 	"\x0eExecuteRequest\x12\x1d\n" +
 	"\n" +
 	"store_name\x18\x01 \x01(\tR\tstoreName\x12\x14\n" +
 	"\x05index\x18\x02 \x01(\x04R\x05index\x12*\n" +
-	"\acommand\x18\x03 \x01(\v2\x10.weir.v1.CommandR\acommand\"M\n" +
+	"\acommand\x18\x03 \x01(\v2\x10.weir.v1.CommandR\acommand\x12%\n" +
+	"\x0etimeout_micros\x18\x04 \x01(\x04R\rtimeoutMicros\"M\n" +
 	"\x0fExecuteResponse\x12\x14\n" +
 	"\x05index\x18\x01 \x01(\x04R\x05index\x12$\n" +
 	"\x05event\x18\x02 \x01(\v2\x0e.weir.v1.EventR\x05event\"\xa6\x01\n" +
diff --git a/api/weir/v1/store.proto b/api/weir/v1/store.proto
index 5e03ddf..611a674 100644
--- a/api/weir/v1/store.proto
+++ b/api/weir/v1/store.proto
@@ -12,8 +12,10 @@ option go_package = "github.com/batchstream/weir-protocol/api/weir/v1;weirv1";
 service StoreService {
   // Execute accepts requests and emits indexed results incrementally.
   // A stream selects one Store and operation kind. Each Read or Mutate request
-  // carries one record with a consecutive one-based index. It is validated
-  // before execution; earlier records may have effects before a later one fails.
+  // carries one record with a consecutive one-based index and receives exactly
+  // one result, in completion order. A record the server cannot admit receives
+  // its own failed result; only an invalid request envelope or sequence ends
+  // the stream, and earlier records may have effects when it does.
   // Mutations to the same resource preserve input order across requests, including
   // after item failures. Different resources may execute concurrently. A stream
   // is not a transaction; ordering across streams follows database semantics.
@@ -34,11 +36,18 @@ message ExecuteRequest {
 
   // Required operation; its kind must remain identical throughout the stream.
   Command command = 3;
+
+  // Optional record deadline in microseconds after the server receives the
+  // request. A record still queued when it expires is NOT_STARTED with
+  // DEADLINE_EXCEEDED; an executing record reports the evidence its
+  // cancellation leaves. Zero sets no record deadline. Native requires zero
+  // and uses the RPC deadline.
+  uint64 timeout_micros = 4;
 }
 
 // ExecuteResponse carries one result or streaming event.
 message ExecuteResponse {
-  // One-based input record ordinal; record results arrive in input order.
+  // One-based input record ordinal; record results arrive in completion order.
   // Every input record has exactly one result. Native events all use 1.
   uint64 index = 1;
 
diff --git a/api/weir/v1/store_grpc.pb.go b/api/weir/v1/store_grpc.pb.go
index c7b5a45..a154b1f 100644
--- a/api/weir/v1/store_grpc.pb.go
+++ b/api/weir/v1/store_grpc.pb.go
@@ -29,8 +29,10 @@ const (
 type StoreServiceClient interface {
 	// Execute accepts requests and emits indexed results incrementally.
 	// A stream selects one Store and operation kind. Each Read or Mutate request
-	// carries one record with a consecutive one-based index. It is validated
-	// before execution; earlier records may have effects before a later one fails.
+	// carries one record with a consecutive one-based index and receives exactly
+	// one result, in completion order. A record the server cannot admit receives
+	// its own failed result; only an invalid request envelope or sequence ends
+	// the stream, and earlier records may have effects when it does.
 	// Mutations to the same resource preserve input order across requests, including
 	// after item failures. Different resources may execute concurrently. A stream
 	// is not a transaction; ordering across streams follows database semantics.
@@ -66,8 +68,10 @@ func (c *storeServiceClient) Execute(ctx context.Context, opts ...grpc.CallOptio
 type StoreServiceServer interface {
 	// Execute accepts requests and emits indexed results incrementally.
 	// A stream selects one Store and operation kind. Each Read or Mutate request
-	// carries one record with a consecutive one-based index. It is validated
-	// before execution; earlier records may have effects before a later one fails.
+	// carries one record with a consecutive one-based index and receives exactly
+	// one result, in completion order. A record the server cannot admit receives
+	// its own failed result; only an invalid request envelope or sequence ends
+	// the stream, and earlier records may have effects when it does.
 	// Mutations to the same resource preserve input order across requests, including
 	// after item failures. Different resources may execute concurrently. A stream
 	// is not a transaction; ordering across streams follows database semantics.

From 47d2cd061b23eab6211ca9489af6857da1f27245 Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Fri, 30 Oct 2026 13:49:53 +0800
Subject: [PATCH 3/6] =?UTF-8?q?=F0=9F=90=9B=20fix:=20round=20TimeoutMicros?=
 =?UTF-8?q?=20up=20without=20overflowing=20near=20its=20limit?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 api/protocol/protocol.go      |  7 +++++--
 api/protocol/protocol_test.go | 17 +++++++++++++++--
 2 files changed, 20 insertions(+), 4 deletions(-)

diff --git a/api/protocol/protocol.go b/api/protocol/protocol.go
index 6368c52..df2fe0f 100644
--- a/api/protocol/protocol.go
+++ b/api/protocol/protocol.go
@@ -239,8 +239,11 @@ func TimeoutMicros(remaining time.Duration) uint64 {
 	if remaining <= time.Microsecond {
 		return 1
 	}
-	micros := (remaining + time.Microsecond - 1) / time.Microsecond
-	return uint64(micros)
+	micros := uint64(remaining / time.Microsecond)
+	if remaining%time.Microsecond != 0 {
+		micros++
+	}
+	return micros
 }
 
 // ValidateExecuteResponse validates one indexed event and rejects unknown fields
diff --git a/api/protocol/protocol_test.go b/api/protocol/protocol_test.go
index 213158d..96b7e99 100644
--- a/api/protocol/protocol_test.go
+++ b/api/protocol/protocol_test.go
@@ -235,13 +235,26 @@ func TestRecordTimeoutNeverShortensOrDisablesTheCallerBudget(t *testing.T) {
 		{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)
 		}
 	}
-	if RecordTimeout(0) != 0 || RecordTimeout(3_000_000) != 3*time.Second || RecordTimeout(math.MaxUint64) != math.MaxInt64 {
-		t.Fatal("record timeout conversion changed")
+	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)
+		}
 	}
 }

From 0fc9878a4f64158ee866c84e52ef6c22306ce0c8 Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Fri, 30 Oct 2026 13:49:53 +0800
Subject: [PATCH 4/6] =?UTF-8?q?=E2=9C=A8=20feat:=20let=20a=20cancel=20requ?=
 =?UTF-8?q?est=20release=20one=20record=20of=20a=20shared=20stream?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A request carrying only store_name and cancel_index cancels an earlier
record without ending the stream, so a stream shared by several callers
can release the work of one caller that stops waiting.

Also state which requests end a stream, that results are at most one
per index, and that record timeouts start when the server reads the
request.
---
 README.md                     | 36 +++++++++++++++++++++++----------
 api/protocol/protocol.go      | 20 +++++++++++++++---
 api/protocol/protocol_test.go | 25 +++++++++++++++++++++++
 api/weir/v1/contract_test.go  |  2 +-
 api/weir/v1/store.pb.go       | 36 +++++++++++++++++++++++++--------
 api/weir/v1/store.proto       | 38 ++++++++++++++++++++++++++---------
 api/weir/v1/store_grpc.pb.go  | 26 ++++++++++++++++--------
 7 files changed, 142 insertions(+), 41 deletions(-)

diff --git a/README.md b/README.md
index 230cdd9..c796405 100644
--- a/README.md
+++ b/README.md
@@ -37,20 +37,34 @@ selected adapter.
 
 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 and the stream continues. An invalid envelope or index
-sequence ends the stream; earlier records may have effects when it does. 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 completion order, so consumers match results by index. One stream can
+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.
-
-`timeout_micros` optionally bounds one record from the moment the server
-receives it. A record still queued when it expires is NOT_STARTED with
-DEADLINE_EXCEEDED; an executing record reports the evidence its cancellation
-leaves. Zero sets no record deadline. Native requires zero and uses the RPC
+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 when it
+was still queued. Cancel requests let a stream shared by several callers
+release the work of one caller who stops waiting. Native streams use RPC
+cancellation instead.
+
 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
diff --git a/api/protocol/protocol.go b/api/protocol/protocol.go
index df2fe0f..1e89d3b 100644
--- a/api/protocol/protocol.go
+++ b/api/protocol/protocol.go
@@ -198,9 +198,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")
 	}
@@ -223,6 +227,16 @@ func ValidateExecuteRequest(req *pb.ExecuteRequest) error {
 	return nil
 }
 
+func validateCancelRequest(req *pb.ExecuteRequest) error {
+	if !ValidStoreName(req.StoreName) || 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 {
@@ -247,7 +261,7 @@ func TimeoutMicros(remaining time.Duration) uint64 {
 }
 
 // ValidateExecuteResponse validates one indexed event and rejects unknown fields
-// recursively. Stream consumers enforce the expected event kind, exactly one
+// 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 {
diff --git a/api/protocol/protocol_test.go b/api/protocol/protocol_test.go
index 96b7e99..d019637 100644
--- a/api/protocol/protocol_test.go
+++ b/api/protocol/protocol_test.go
@@ -115,6 +115,31 @@ 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)
+	}
+	document := &pb.Document{ContentType: "application/json", Data: []byte(`{}`)}
+	action := &pb.MutateRequest_Put{Put: document}
+	mutation := &pb.MutateRequest{Resource: "db/c/s:a", Action: action}
+	operation := &pb.Command_Mutate{Mutate: mutation}
+	command := &pb.Command{Operation: operation}
+	for _, mutate := range []func(*pb.ExecuteRequest){
+		func(r *pb.ExecuteRequest) { r.StoreName = "" },
+		func(r *pb.ExecuteRequest) { r.Index = 3 },
+		func(r *pb.ExecuteRequest) { r.Command = command },
+		func(r *pb.ExecuteRequest) { r.TimeoutMicros = 1 },
+		func(r *pb.ExecuteRequest) { r.ProtoReflect().SetUnknown([]byte{0xf8, 0x07, 1}) },
+	} {
+		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}
diff --git a/api/weir/v1/contract_test.go b/api/weir/v1/contract_test.go
index fb4dff1..8d5754a 100644
--- a/api/weir/v1/contract_test.go
+++ b/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"}, {name: "timeout_micros", kind: protoreflect.Uint64Kind}}
+	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)
diff --git a/api/weir/v1/store.pb.go b/api/weir/v1/store.pb.go
index 673f076..17edeb3 100644
--- a/api/weir/v1/store.pb.go
+++ b/api/weir/v1/store.pb.go
@@ -297,12 +297,23 @@ type ExecuteRequest struct {
 	Index uint64 `protobuf:"varint,2,opt,name=index,proto3" json:"index,omitempty"`
 	// Required operation; its kind must remain identical throughout the stream.
 	Command *Command `protobuf:"bytes,3,opt,name=command,proto3" json:"command,omitempty"`
-	// Optional record deadline in microseconds after the server receives the
-	// request. A record still queued when it expires is NOT_STARTED with
-	// DEADLINE_EXCEEDED; an executing record reports the evidence its
-	// cancellation leaves. Zero sets no record deadline. Native requires zero
-	// and uses the RPC deadline.
+	// Optional Read or Mutate deadline in microseconds, measured from when the
+	// server reads the request. Time spent in transport buffers before then 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.
 	TimeoutMicros uint64 `protobuf:"varint,4,opt,name=timeout_micros,json=timeoutMicros,proto3" json:"timeout_micros,omitempty"`
+	// Cancels the earlier record of this stream with this index instead of
+	// carrying one. A cancel request sets only store_name and cancel_index, does
+	// not advance the index and receives no response; the canceled record still
+	// receives its own result. A queued record fails with CANCELLED, NOT_STARTED
+	// for a mutation; an executing record reports the evidence its cancellation
+	// leaves, and a completed record is unaffected. The server reads requests in
+	// order, so a cancel request takes effect once it is read. Native streams
+	// use RPC cancellation instead.
+	CancelIndex   uint64 `protobuf:"varint,5,opt,name=cancel_index,json=cancelIndex,proto3" json:"cancel_index,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -365,11 +376,19 @@ func (x *ExecuteRequest) GetTimeoutMicros() uint64 {
 	return 0
 }
 
+func (x *ExecuteRequest) GetCancelIndex() uint64 {
+	if x != nil {
+		return x.CancelIndex
+	}
+	return 0
+}
+
 // ExecuteResponse carries one result or streaming event.
 type ExecuteResponse struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// One-based input record ordinal; record results arrive in completion order.
-	// Every input record has exactly one result. Native events all use 1.
+	// An index receives at most one result, and every record receives one
+	// unless the RPC fails first. Native events all use 1.
 	Index uint64 `protobuf:"varint,1,opt,name=index,proto3" json:"index,omitempty"`
 	// Required typed event; it must match the stream's operation kind.
 	Event         *Event `protobuf:"bytes,2,opt,name=event,proto3" json:"event,omitempty"`
@@ -1546,13 +1565,14 @@ var File_api_weir_v1_store_proto protoreflect.FileDescriptor
 
 const file_api_weir_v1_store_proto_rawDesc = "" +
 	"\n" +
-	"\x17api/weir/v1/store.proto\x12\aweir.v1\"\x98\x01\n" +
+	"\x17api/weir/v1/store.proto\x12\aweir.v1\"\xbb\x01\n" +
 	"\x0eExecuteRequest\x12\x1d\n" +
 	"\n" +
 	"store_name\x18\x01 \x01(\tR\tstoreName\x12\x14\n" +
 	"\x05index\x18\x02 \x01(\x04R\x05index\x12*\n" +
 	"\acommand\x18\x03 \x01(\v2\x10.weir.v1.CommandR\acommand\x12%\n" +
-	"\x0etimeout_micros\x18\x04 \x01(\x04R\rtimeoutMicros\"M\n" +
+	"\x0etimeout_micros\x18\x04 \x01(\x04R\rtimeoutMicros\x12!\n" +
+	"\fcancel_index\x18\x05 \x01(\x04R\vcancelIndex\"M\n" +
 	"\x0fExecuteResponse\x12\x14\n" +
 	"\x05index\x18\x01 \x01(\x04R\x05index\x12$\n" +
 	"\x05event\x18\x02 \x01(\v2\x0e.weir.v1.EventR\x05event\"\xa6\x01\n" +
diff --git a/api/weir/v1/store.proto b/api/weir/v1/store.proto
index 611a674..edd1e2a 100644
--- a/api/weir/v1/store.proto
+++ b/api/weir/v1/store.proto
@@ -12,10 +12,15 @@ option go_package = "github.com/batchstream/weir-protocol/api/weir/v1;weirv1";
 service StoreService {
   // Execute accepts requests and emits indexed results incrementally.
   // A stream selects one Store and operation kind. Each Read or Mutate request
-  // carries one record with a consecutive one-based index and receives exactly
-  // one result, in completion order. A record the server cannot admit receives
-  // its own failed result; only an invalid request envelope or sequence ends
-  // the stream, and earlier records may have effects when it does.
+  // carries one record with a consecutive one-based index and receives at most
+  // one result, in completion order; consumers match results by index. A
+  // record the server cannot admit, such as one its adapter rejects, receives
+  // its own failed result, NOT_STARTED for a mutation, and the stream
+  // continues. A request that fails validation, 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. A
+  // cancel request cancels one earlier record without ending the stream.
   // Mutations to the same resource preserve input order across requests, including
   // after item failures. Different resources may execute concurrently. A stream
   // is not a transaction; ordering across streams follows database semantics.
@@ -37,18 +42,31 @@ message ExecuteRequest {
   // Required operation; its kind must remain identical throughout the stream.
   Command command = 3;
 
-  // Optional record deadline in microseconds after the server receives the
-  // request. A record still queued when it expires is NOT_STARTED with
-  // DEADLINE_EXCEEDED; an executing record reports the evidence its
-  // cancellation leaves. Zero sets no record deadline. Native requires zero
-  // and uses the RPC deadline.
+  // Optional Read or Mutate deadline in microseconds, measured from when the
+  // server reads the request. Time spent in transport buffers before then 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.
   uint64 timeout_micros = 4;
+
+  // Cancels the earlier record of this stream with this index instead of
+  // carrying one. A cancel request sets only store_name and cancel_index, does
+  // not advance the index and receives no response; the canceled record still
+  // receives its own result. A queued record fails with CANCELLED, NOT_STARTED
+  // for a mutation; an executing record reports the evidence its cancellation
+  // leaves, and a completed record is unaffected. The server reads requests in
+  // order, so a cancel request takes effect once it is read. Native streams
+  // use RPC cancellation instead.
+  uint64 cancel_index = 5;
 }
 
 // ExecuteResponse carries one result or streaming event.
 message ExecuteResponse {
   // One-based input record ordinal; record results arrive in completion order.
-  // Every input record has exactly one result. Native events all use 1.
+  // An index receives at most one result, and every record receives one
+  // unless the RPC fails first. Native events all use 1.
   uint64 index = 1;
 
   // Required typed event; it must match the stream's operation kind.
diff --git a/api/weir/v1/store_grpc.pb.go b/api/weir/v1/store_grpc.pb.go
index a154b1f..6a780c1 100644
--- a/api/weir/v1/store_grpc.pb.go
+++ b/api/weir/v1/store_grpc.pb.go
@@ -29,10 +29,15 @@ const (
 type StoreServiceClient interface {
 	// Execute accepts requests and emits indexed results incrementally.
 	// A stream selects one Store and operation kind. Each Read or Mutate request
-	// carries one record with a consecutive one-based index and receives exactly
-	// one result, in completion order. A record the server cannot admit receives
-	// its own failed result; only an invalid request envelope or sequence ends
-	// the stream, and earlier records may have effects when it does.
+	// carries one record with a consecutive one-based index and receives at most
+	// one result, in completion order; consumers match results by index. A
+	// record the server cannot admit, such as one its adapter rejects, receives
+	// its own failed result, NOT_STARTED for a mutation, and the stream
+	// continues. A request that fails validation, 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. A
+	// cancel request cancels one earlier record without ending the stream.
 	// Mutations to the same resource preserve input order across requests, including
 	// after item failures. Different resources may execute concurrently. A stream
 	// is not a transaction; ordering across streams follows database semantics.
@@ -68,10 +73,15 @@ func (c *storeServiceClient) Execute(ctx context.Context, opts ...grpc.CallOptio
 type StoreServiceServer interface {
 	// Execute accepts requests and emits indexed results incrementally.
 	// A stream selects one Store and operation kind. Each Read or Mutate request
-	// carries one record with a consecutive one-based index and receives exactly
-	// one result, in completion order. A record the server cannot admit receives
-	// its own failed result; only an invalid request envelope or sequence ends
-	// the stream, and earlier records may have effects when it does.
+	// carries one record with a consecutive one-based index and receives at most
+	// one result, in completion order; consumers match results by index. A
+	// record the server cannot admit, such as one its adapter rejects, receives
+	// its own failed result, NOT_STARTED for a mutation, and the stream
+	// continues. A request that fails validation, 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. A
+	// cancel request cancels one earlier record without ending the stream.
 	// Mutations to the same resource preserve input order across requests, including
 	// after item failures. Different resources may execute concurrently. A stream
 	// is not a transaction; ordering across streams follows database semantics.

From 6c8e472bbef77ec202729980fcbb9ac07ce008dd Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Fri, 30 Oct 2026 14:05:27 +0800
Subject: [PATCH 5/6] =?UTF-8?q?=E2=9C=A8=20feat:=20promise=20a=2032-record?=
 =?UTF-8?q?=20read-ahead=20window=20for=20cancel=20requests?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A server reads up to RecordWindow records without results per stream,
whatever its Store queue holds, so a client within the window never
waits to send a cancel request. Also describe cancel requests in the
ExecuteRequest fields, reject cancel_index UINT64_MAX, and test that
every other field stays unset in a cancel request.
---
 README.md                     | 25 +++++++++++++++++--------
 api/protocol/protocol.go      |  7 ++++++-
 api/protocol/protocol_test.go | 35 +++++++++++++++++++++++++----------
 api/weir/v1/store.pb.go       | 15 +++++++++------
 api/weir/v1/store.proto       | 19 +++++++++++++------
 api/weir/v1/store_grpc.pb.go  |  8 ++++++++
 6 files changed, 78 insertions(+), 31 deletions(-)

diff --git a/README.md b/README.md
index c796405..7b2d0b1 100644
--- a/README.md
+++ b/README.md
@@ -54,16 +54,25 @@ caller on it.
 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.
+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 when it
-was still queued. Cancel requests let a stream shared by several callers
-release the work of one caller who stops waiting. Native streams use RPC
-cancellation instead.
+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.
+
+A server reads ahead of its results for up to `RecordWindow` (32) records
+without results per stream, whatever its Store queue holds. A client that never
+has more than 32 records without results never waits to send a record or cancel
+request, 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
diff --git a/api/protocol/protocol.go b/api/protocol/protocol.go
index 1e89d3b..32e2d41 100644
--- a/api/protocol/protocol.go
+++ b/api/protocol/protocol.go
@@ -21,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 a server reads ahead
+// of its results on one stream. A client that never exceeds it never waits to
+// send a record or cancel request.
+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!#$&^_.+-]+$`)
 
@@ -228,7 +233,7 @@ func ValidateExecuteRequest(req *pb.ExecuteRequest) error {
 }
 
 func validateCancelRequest(req *pb.ExecuteRequest) error {
-	if !ValidStoreName(req.StoreName) || req.Index != 0 || req.Command != nil || req.TimeoutMicros != 0 {
+	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()) {
diff --git a/api/protocol/protocol_test.go b/api/protocol/protocol_test.go
index d019637..e175983 100644
--- a/api/protocol/protocol_test.go
+++ b/api/protocol/protocol_test.go
@@ -120,18 +120,33 @@ func TestCancelRequestCarriesOnlyItsStoreAndIndex(t *testing.T) {
 	if err := ValidateExecuteRequest(request); err != nil {
 		t.Fatal("cancel request rejected", err)
 	}
-	document := &pb.Document{ContentType: "application/json", Data: []byte(`{}`)}
-	action := &pb.MutateRequest_Put{Put: document}
-	mutation := &pb.MutateRequest{Resource: "db/c/s:a", Action: action}
-	operation := &pb.Command_Mutate{Mutate: mutation}
-	command := &pb.Command{Operation: operation}
-	for _, mutate := range []func(*pb.ExecuteRequest){
+	invalid := []func(*pb.ExecuteRequest){
 		func(r *pb.ExecuteRequest) { r.StoreName = "" },
-		func(r *pb.ExecuteRequest) { r.Index = 3 },
-		func(r *pb.ExecuteRequest) { r.Command = command },
-		func(r *pb.ExecuteRequest) { r.TimeoutMicros = 1 },
+		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 {
diff --git a/api/weir/v1/store.pb.go b/api/weir/v1/store.pb.go
index 17edeb3..9f69e4e 100644
--- a/api/weir/v1/store.pb.go
+++ b/api/weir/v1/store.pb.go
@@ -287,15 +287,18 @@ func (NativeCompletion) EnumDescriptor() ([]byte, []int) {
 	return file_api_weir_v1_store_proto_rawDescGZIP(), []int{3}
 }
 
-// ExecuteRequest binds one operation to its Store and index.
+// ExecuteRequest carries one operation with its Store and index, or cancels an
+// earlier record of the stream.
 type ExecuteRequest struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// Store that owns all resources; must remain identical throughout the stream.
 	StoreName string `protobuf:"bytes,1,opt,name=store_name,json=storeName,proto3" json:"store_name,omitempty"`
 	// Consecutive one-based record ordinal. Each Read or Mutate request advances
-	// it by one. Native always uses 1. UINT64_MAX is invalid.
+	// it by one. Native always uses 1. UINT64_MAX is invalid. Zero in a cancel
+	// request.
 	Index uint64 `protobuf:"varint,2,opt,name=index,proto3" json:"index,omitempty"`
-	// Required operation; its kind must remain identical throughout the stream.
+	// Operation, required except in a cancel request; its kind must remain
+	// identical throughout the stream.
 	Command *Command `protobuf:"bytes,3,opt,name=command,proto3" json:"command,omitempty"`
 	// Optional Read or Mutate deadline in microseconds, measured from when the
 	// server reads the request. Time spent in transport buffers before then is
@@ -310,9 +313,9 @@ type ExecuteRequest struct {
 	// not advance the index and receives no response; the canceled record still
 	// receives its own result. A queued record fails with CANCELLED, NOT_STARTED
 	// for a mutation; an executing record reports the evidence its cancellation
-	// leaves, and a completed record is unaffected. The server reads requests in
-	// order, so a cancel request takes effect once it is read. Native streams
-	// use RPC cancellation instead.
+	// leaves, and a completed record is unaffected. A cancel request takes effect
+	// once the server reads it, at the latest when the stream's window has room.
+	// UINT64_MAX is invalid. Native streams use RPC cancellation instead.
 	CancelIndex   uint64 `protobuf:"varint,5,opt,name=cancel_index,json=cancelIndex,proto3" json:"cancel_index,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
diff --git a/api/weir/v1/store.proto b/api/weir/v1/store.proto
index edd1e2a..aafcf49 100644
--- a/api/weir/v1/store.proto
+++ b/api/weir/v1/store.proto
@@ -21,6 +21,10 @@ service StoreService {
   // 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. A
   // cancel request cancels one earlier record without ending the stream.
+  // A server reads ahead of its results for up to 32 records without results
+  // per stream, whatever its Store queue holds. A client that never has more
+  // than 32 records without results never waits to send a record or cancel
+  // request; a further record waits until a result frees room.
   // Mutations to the same resource preserve input order across requests, including
   // after item failures. Different resources may execute concurrently. A stream
   // is not a transaction; ordering across streams follows database semantics.
@@ -30,16 +34,19 @@ service StoreService {
   rpc Execute(stream ExecuteRequest) returns (stream ExecuteResponse);
 }
 
-// ExecuteRequest binds one operation to its Store and index.
+// ExecuteRequest carries one operation with its Store and index, or cancels an
+// earlier record of the stream.
 message ExecuteRequest {
   // Store that owns all resources; must remain identical throughout the stream.
   string store_name = 1;
 
   // Consecutive one-based record ordinal. Each Read or Mutate request advances
-  // it by one. Native always uses 1. UINT64_MAX is invalid.
+  // it by one. Native always uses 1. UINT64_MAX is invalid. Zero in a cancel
+  // request.
   uint64 index = 2;
 
-  // Required operation; its kind must remain identical throughout the stream.
+  // Operation, required except in a cancel request; its kind must remain
+  // identical throughout the stream.
   Command command = 3;
 
   // Optional Read or Mutate deadline in microseconds, measured from when the
@@ -56,9 +63,9 @@ message ExecuteRequest {
   // not advance the index and receives no response; the canceled record still
   // receives its own result. A queued record fails with CANCELLED, NOT_STARTED
   // for a mutation; an executing record reports the evidence its cancellation
-  // leaves, and a completed record is unaffected. The server reads requests in
-  // order, so a cancel request takes effect once it is read. Native streams
-  // use RPC cancellation instead.
+  // leaves, and a completed record is unaffected. A cancel request takes effect
+  // once the server reads it, at the latest when the stream's window has room.
+  // UINT64_MAX is invalid. Native streams use RPC cancellation instead.
   uint64 cancel_index = 5;
 }
 
diff --git a/api/weir/v1/store_grpc.pb.go b/api/weir/v1/store_grpc.pb.go
index 6a780c1..dbf80bc 100644
--- a/api/weir/v1/store_grpc.pb.go
+++ b/api/weir/v1/store_grpc.pb.go
@@ -38,6 +38,10 @@ type StoreServiceClient interface {
 	// 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. A
 	// cancel request cancels one earlier record without ending the stream.
+	// A server reads ahead of its results for up to 32 records without results
+	// per stream, whatever its Store queue holds. A client that never has more
+	// than 32 records without results never waits to send a record or cancel
+	// request; a further record waits until a result frees room.
 	// Mutations to the same resource preserve input order across requests, including
 	// after item failures. Different resources may execute concurrently. A stream
 	// is not a transaction; ordering across streams follows database semantics.
@@ -82,6 +86,10 @@ type StoreServiceServer interface {
 	// 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. A
 	// cancel request cancels one earlier record without ending the stream.
+	// A server reads ahead of its results for up to 32 records without results
+	// per stream, whatever its Store queue holds. A client that never has more
+	// than 32 records without results never waits to send a record or cancel
+	// request; a further record waits until a result frees room.
 	// Mutations to the same resource preserve input order across requests, including
 	// after item failures. Different resources may execute concurrently. A stream
 	// is not a transaction; ordering across streams follows database semantics.

From 1ff630340c6c9e615f7e83a99d490e0664be7759 Mon Sep 17 00:00:00 2001
From: Randy <857271370@qq.com>
Date: Fri, 30 Oct 2026 14:19:06 +0800
Subject: [PATCH 6/6] =?UTF-8?q?=F0=9F=93=9D=20docs:=20state=20the=20record?=
 =?UTF-8?q?=20window=20as=20room=20each=20stream=20is=20guaranteed?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 README.md                    |  9 ++++-----
 api/protocol/protocol.go     |  6 +++---
 api/weir/v1/store.pb.go      |  4 ++--
 api/weir/v1/store.proto      | 12 ++++++------
 api/weir/v1/store_grpc.pb.go | 16 ++++++++--------
 5 files changed, 23 insertions(+), 24 deletions(-)

diff --git a/README.md b/README.md
index 7b2d0b1..b8b8f84 100644
--- a/README.md
+++ b/README.md
@@ -68,11 +68,10 @@ 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.
 
-A server reads ahead of its results for up to `RecordWindow` (32) records
-without results per stream, whatever its Store queue holds. A client that never
-has more than 32 records without results never waits to send a record or cancel
-request, so its cancel requests take effect as soon as they arrive; a further
-record waits until a result frees room.
+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
diff --git a/api/protocol/protocol.go b/api/protocol/protocol.go
index 32e2d41..2250d52 100644
--- a/api/protocol/protocol.go
+++ b/api/protocol/protocol.go
@@ -21,9 +21,9 @@ 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 a server reads ahead
-// of its results on one stream. A client that never exceeds it never waits to
-// send a record or cancel request.
+// 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])*$`)
diff --git a/api/weir/v1/store.pb.go b/api/weir/v1/store.pb.go
index 9f69e4e..018038f 100644
--- a/api/weir/v1/store.pb.go
+++ b/api/weir/v1/store.pb.go
@@ -314,8 +314,8 @@ type ExecuteRequest struct {
 	// receives its own result. A queued record fails with CANCELLED, NOT_STARTED
 	// for a mutation; an executing record reports the evidence its cancellation
 	// leaves, and a completed record is unaffected. A cancel request takes effect
-	// once the server reads it, at the latest when the stream's window has room.
-	// UINT64_MAX is invalid. Native streams use RPC cancellation instead.
+	// once the server reads it, which a record beyond the stream's window can
+	// delay. UINT64_MAX is invalid. Native streams use RPC cancellation instead.
 	CancelIndex   uint64 `protobuf:"varint,5,opt,name=cancel_index,json=cancelIndex,proto3" json:"cancel_index,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
diff --git a/api/weir/v1/store.proto b/api/weir/v1/store.proto
index aafcf49..22ea313 100644
--- a/api/weir/v1/store.proto
+++ b/api/weir/v1/store.proto
@@ -21,10 +21,10 @@ service StoreService {
   // 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. A
   // cancel request cancels one earlier record without ending the stream.
-  // A server reads ahead of its results for up to 32 records without results
-  // per stream, whatever its Store queue holds. A client that never has more
-  // than 32 records without results never waits to send a record or cancel
-  // request; a further record waits until a result frees room.
+  // Each stream has room for 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 once they arrive; a further record waits until a result frees room.
   // Mutations to the same resource preserve input order across requests, including
   // after item failures. Different resources may execute concurrently. A stream
   // is not a transaction; ordering across streams follows database semantics.
@@ -64,8 +64,8 @@ message ExecuteRequest {
   // receives its own result. A queued record fails with CANCELLED, NOT_STARTED
   // for a mutation; an executing record reports the evidence its cancellation
   // leaves, and a completed record is unaffected. A cancel request takes effect
-  // once the server reads it, at the latest when the stream's window has room.
-  // UINT64_MAX is invalid. Native streams use RPC cancellation instead.
+  // once the server reads it, which a record beyond the stream's window can
+  // delay. UINT64_MAX is invalid. Native streams use RPC cancellation instead.
   uint64 cancel_index = 5;
 }
 
diff --git a/api/weir/v1/store_grpc.pb.go b/api/weir/v1/store_grpc.pb.go
index dbf80bc..0c54a6e 100644
--- a/api/weir/v1/store_grpc.pb.go
+++ b/api/weir/v1/store_grpc.pb.go
@@ -38,10 +38,10 @@ type StoreServiceClient interface {
 	// 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. A
 	// cancel request cancels one earlier record without ending the stream.
-	// A server reads ahead of its results for up to 32 records without results
-	// per stream, whatever its Store queue holds. A client that never has more
-	// than 32 records without results never waits to send a record or cancel
-	// request; a further record waits until a result frees room.
+	// Each stream has room for 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 once they arrive; a further record waits until a result frees room.
 	// Mutations to the same resource preserve input order across requests, including
 	// after item failures. Different resources may execute concurrently. A stream
 	// is not a transaction; ordering across streams follows database semantics.
@@ -86,10 +86,10 @@ type StoreServiceServer interface {
 	// 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. A
 	// cancel request cancels one earlier record without ending the stream.
-	// A server reads ahead of its results for up to 32 records without results
-	// per stream, whatever its Store queue holds. A client that never has more
-	// than 32 records without results never waits to send a record or cancel
-	// request; a further record waits until a result frees room.
+	// Each stream has room for 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 once they arrive; a further record waits until a result frees room.
 	// Mutations to the same resource preserve input order across requests, including
 	// after item failures. Different resources may execute concurrently. A stream
 	// is not a transaction; ordering across streams follows database semantics.