Let one Execute message carry several requests or responses

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.16.0
+go get github.com/batchstream/weir-protocol@v0.17.0
 ```
 
 The dependency direction is one-way: Weir and the Go SDK depend on this module.
@@ -27,13 +27,19 @@ Mutate request carries one record. Servers aggregate admitted records into
 database batches; clients send and consume incrementally without collecting a
 wire batch. The protocol imposes no item or payload-size limit on business calls
 or individual records; servers may bound the size of one message. Backend
-protocols govern accepted documents and commands; gRPC carries one request or
-event per message. Native carries a required opaque Document. NativeHead carries
-optional opaque metadata and a content type for subsequent body chunks. Request
-and metadata bytes are opaque to the public protocol. Their content types select
-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.
+protocols govern accepted documents and commands. Native carries a required
+opaque Document. NativeHead carries optional opaque metadata and a content type
+for subsequent body chunks. Request and metadata bytes are opaque to the public
+protocol. Their content types select 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 gRPC message, `ExecuteRequests` or `ExecuteResponses`, carries one or more
+requests or responses in stream order. A peer handles each exactly as if it came
+in a message of its own, so a sender packs everything it has ready into one
+message instead of paying for a message per record. `ValidateExecuteRequests`
+and `ValidateExecuteResponses` check a message's own envelope; its requests and
+responses are validated one at a time as they are applied.
 
 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

api/protocol/protocol.go

@@ -203,6 +203,27 @@ func NativeFailure(started bool, failure *pb.Failure) *pb.NativeEnd {
 	return end
 }
 
+// ValidateExecuteRequests checks one request message's own envelope: at least
+// one request and no unknown fields of its own. Consumers validate and apply
+// its requests in order with ValidateExecuteRequest, exactly as if each came in
+// a message of its own.
+func ValidateExecuteRequests(message *pb.ExecuteRequests) error {
+	if message == nil || len(message.Requests) == 0 || len(message.ProtoReflect().GetUnknown()) != 0 {
+		return fmt.Errorf("invalid Execute request message")
+	}
+	return nil
+}
+
+// ValidateExecuteResponses checks one response message's own envelope: at
+// least one response and no unknown fields of its own. Consumers validate and
+// apply its responses in order with ValidateExecuteResponse.
+func ValidateExecuteResponses(message *pb.ExecuteResponses) error {
+	if message == nil || len(message.Responses) == 0 || len(message.ProtoReflect().GetUnknown()) != 0 {
+		return fmt.Errorf("invalid Execute response message")
+	}
+	return nil
+}
+
 // 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.

api/protocol/protocol_test.go

@@ -115,6 +115,39 @@ func TestExecuteTypedEnvelopeAndUnknownFields(t *testing.T) {
 	}
 }
 
+func TestMessagesCarryAtLeastOneRequestOrResponse(t *testing.T) {
+	request := &pb.ExecuteRequest{StoreName: "records", CancelIndex: 1}
+	requests := &pb.ExecuteRequests{Requests: []*pb.ExecuteRequest{request}}
+	if err := ValidateExecuteRequests(requests); err != nil {
+		t.Fatal(err)
+	}
+	// Requests are validated one at a time as they are applied.
+	invalid := &pb.ExecuteRequest{}
+	requests.Requests = append(requests.Requests, invalid)
+	if err := ValidateExecuteRequests(requests); err != nil {
+		t.Fatal(err)
+	}
+	requests.ProtoReflect().SetUnknown([]byte{0xf8, 0x07, 1})
+	empty := &pb.ExecuteRequests{}
+	for _, message := range []*pb.ExecuteRequests{nil, empty, requests} {
+		if err := ValidateExecuteRequests(message); err == nil {
+			t.Fatal("accepted invalid request message", message)
+		}
+	}
+	response := &pb.ExecuteResponse{Index: 1}
+	responses := &pb.ExecuteResponses{Responses: []*pb.ExecuteResponse{response}}
+	if err := ValidateExecuteResponses(responses); err != nil {
+		t.Fatal(err)
+	}
+	responses.ProtoReflect().SetUnknown([]byte{0xf8, 0x07, 1})
+	emptyResponses := &pb.ExecuteResponses{}
+	for _, message := range []*pb.ExecuteResponses{nil, emptyResponses, responses} {
+		if err := ValidateExecuteResponses(message); err == nil {
+			t.Fatal("accepted invalid response message", message)
+		}
+	}
+}
+
 func TestCancelRequestCarriesOnlyItsStoreAndIndex(t *testing.T) {
 	request := &pb.ExecuteRequest{StoreName: "records", CancelIndex: 3}
 	if err := ValidateExecuteRequest(request); err != nil {

api/weir/v1/contract_test.go

@@ -55,16 +55,20 @@ func TestPublicStoreContract(t *testing.T) {
 		t.Fatal("public service must expose exactly one Execute RPC")
 	}
 	execute := service.Methods().ByName("Execute")
-	if execute == nil || execute.Input().FullName() != "weir.v1.ExecuteRequest" || execute.Output().FullName() != "weir.v1.ExecuteResponse" || !execute.IsStreamingClient() || !execute.IsStreamingServer() {
+	if execute == nil || execute.Input().FullName() != "weir.v1.ExecuteRequests" || execute.Output().FullName() != "weir.v1.ExecuteResponses" || !execute.IsStreamingClient() || !execute.IsStreamingServer() {
 		t.Fatal("Execute bidirectional contract changed", execute)
 	}
 	if StoreService_Execute_FullMethodName != "/weir.v1.StoreService/Execute" {
 		t.Fatal("generated public method path changed")
 	}
+	requests := []fieldContract{{name: "requests", kind: protoreflect.MessageKind, message: "weir.v1.ExecuteRequest", repeated: true}}
+	assertFields(t, execute.Input(), requests)
+	responses := []fieldContract{{name: "responses", kind: protoreflect.MessageKind, message: "weir.v1.ExecuteResponse", repeated: true}}
+	assertFields(t, execute.Output(), responses)
 	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)
+	assertFields(t, execute.Input().Fields().ByName("requests").Message(), executeRequest)
 	executeResponse := []fieldContract{{name: "index", kind: protoreflect.Uint64Kind}, {name: "event", kind: protoreflect.MessageKind, message: "weir.v1.Event"}}
-	assertFields(t, execute.Output(), executeResponse)
+	assertFields(t, execute.Output().Fields().ByName("responses").Message(), executeResponse)
 }
 
 func TestPublicCommandAndEventContract(t *testing.T) {
@@ -143,6 +147,15 @@ func TestIndexedExecutionWireEncoding(t *testing.T) {
 	if !bytes.Equal(encoded, wantRequest) {
 		t.Fatal("indexed execution wire changed", encoded)
 	}
+	requests := &ExecuteRequests{Requests: []*ExecuteRequest{request, request}}
+	encoded, err = proto.Marshal(requests)
+	if err != nil {
+		t.Fatal(err)
+	}
+	packedRequest := append([]byte{0x0a, byte(len(wantRequest))}, wantRequest...)
+	if !bytes.Equal(encoded, append(packedRequest, packedRequest...)) {
+		t.Fatal("packed execution wire changed", encoded)
+	}
 	empty := &Empty{}
 	missing := &ReadResult_Missing{Missing: empty}
 	result := &ReadResult{Result: missing}
@@ -157,4 +170,12 @@ func TestIndexedExecutionWireEncoding(t *testing.T) {
 	if !bytes.Equal(encoded, wantResponse) {
 		t.Fatal("indexed per-record result wire changed", encoded)
 	}
+	responses := &ExecuteResponses{Responses: []*ExecuteResponse{response}}
+	encoded, err = proto.Marshal(responses)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(encoded, append([]byte{0x0a, byte(len(wantResponse))}, wantResponse...)) {
+		t.Fatal("packed per-record result wire changed", encoded)
+	}
 }