From f2d89d5b450792b3801a8d638e70358df70da4a0 Mon Sep 17 00:00:00 2001 From: cabbage Date: Sun, 6 Sep 2026 07:35:51 +0000 Subject: [PATCH] fix: carry immutable hash in command dispatch --- gen/go/rvbox/v1/agent.pb.go | 18 ++++++++++++++---- internal/agentproto/validate.go | 2 +- internal/agentproto/validate_test.go | 4 +++- internal/server/session/agent_server.go | 2 +- internal/server/store/dispatch.go | 13 ++++++++----- protos/rvbox/v1/agent.proto | 2 ++ .../clientagent_integration_test.go | 2 +- 7 files changed, 30 insertions(+), 13 deletions(-) diff --git a/gen/go/rvbox/v1/agent.pb.go b/gen/go/rvbox/v1/agent.pb.go index b3fb974..8b096e4 100644 --- a/gen/go/rvbox/v1/agent.pb.go +++ b/gen/go/rvbox/v1/agent.pb.go @@ -534,8 +534,10 @@ type CommandDispatch struct { IssueTime *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=issue_time,json=issueTime,proto3" json:"issue_time,omitempty"` Spec *ExecutionSpec `protobuf:"bytes,5,opt,name=spec,proto3" json:"spec,omitempty"` QueueExpiryTime *timestamppb.Timestamp `protobuf:"bytes,6,opt,name=queue_expiry_time,json=queueExpiryTime,proto3" json:"queue_expiry_time,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Required immutable digest for client-side UUID/hash idempotency checks. + ImmutableRequestSha256 []byte `protobuf:"bytes,7,opt,name=immutable_request_sha256,json=immutableRequestSha256,proto3" json:"immutable_request_sha256,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *CommandDispatch) Reset() { @@ -610,6 +612,13 @@ func (x *CommandDispatch) GetQueueExpiryTime() *timestamppb.Timestamp { return nil } +func (x *CommandDispatch) GetImmutableRequestSha256() []byte { + if x != nil { + return x.ImmutableRequestSha256 + } + return nil +} + type CommandAccepted struct { state protoimpl.MessageState `protogen:"open.v1"` IssueUuid string `protobuf:"bytes,1,opt,name=issue_uuid,json=issueUuid,proto3" json:"issue_uuid,omitempty"` @@ -1502,7 +1511,7 @@ const file_rvbox_v1_agent_proto_rawDesc = "" + "\rServerWelcome\x12F\n" + "\x11selected_protocol\x18\x01 \x01(\v2\x19.rvbox.v1.ProtocolVersionR\x10selectedProtocol\x12;\n" + "\vserver_time\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\n" + - "serverTime\"\xc7\x02\n" + + "serverTime\"\x81\x03\n" + "\x0fCommandDispatch\x12\x1d\n" + "\n" + "issue_uuid\x18\x01 \x01(\tR\tissueUuid\x12)\n" + @@ -1511,7 +1520,8 @@ const file_rvbox_v1_agent_proto_rawDesc = "" + "\n" + "issue_time\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\tissueTime\x12+\n" + "\x04spec\x18\x05 \x01(\v2\x17.rvbox.v1.ExecutionSpecR\x04spec\x12F\n" + - "\x11queue_expiry_time\x18\x06 \x01(\v2\x1a.google.protobuf.TimestampR\x0fqueueExpiryTime\"\xad\x01\n" + + "\x11queue_expiry_time\x18\x06 \x01(\v2\x1a.google.protobuf.TimestampR\x0fqueueExpiryTime\x128\n" + + "\x18immutable_request_sha256\x18\a \x01(\fR\x16immutableRequestSha256\"\xad\x01\n" + "\x0fCommandAccepted\x12\x1d\n" + "\n" + "issue_uuid\x18\x01 \x01(\tR\tissueUuid\x12)\n" + diff --git a/internal/agentproto/validate.go b/internal/agentproto/validate.go index 0f7315e..92a25d4 100644 --- a/internal/agentproto/validate.go +++ b/internal/agentproto/validate.go @@ -188,7 +188,7 @@ func validateClientHello(hello *rvboxv1.ClientHello) error { } func validateCommandDispatch(dispatch *rvboxv1.CommandDispatch, limits Limits, peerPlatform rvboxv1.Platform) error { - if dispatch == nil || dispatch.CommandRevision == 0 || dispatch.TargetSessionGeneration == 0 || dispatch.IssueTime == nil || dispatch.IssueTime.CheckValid() != nil { + if dispatch == nil || dispatch.CommandRevision == 0 || dispatch.TargetSessionGeneration == 0 || dispatch.IssueTime == nil || dispatch.IssueTime.CheckValid() != nil || len(dispatch.ImmutableRequestSha256) != sha256.Size { return fmt.Errorf("%w: invalid command dispatch", ErrInvalidEnvelope) } if err := validateIssueUUID(dispatch.IssueUuid); err != nil { diff --git a/internal/agentproto/validate_test.go b/internal/agentproto/validate_test.go index 20f20be..cc8cd9a 100644 --- a/internal/agentproto/validate_test.go +++ b/internal/agentproto/validate_test.go @@ -2,6 +2,7 @@ package agentproto import ( "bytes" + "crypto/sha256" "errors" "strings" "testing" @@ -140,7 +141,7 @@ func TestWindowsEnvironmentCaseCollision_BH_LAUNCH_01(t *testing.T) { func TestCommandDispatchUUIDRevisionAndTimestamp_BH_IDEM_01(t *testing.T) { t.Parallel() - dispatch := &rvboxv1.CommandDispatch{IssueUuid: testIssueUUID, CommandRevision: 1, TargetSessionGeneration: 2, IssueTime: fixedTimestamp(), Spec: validCommandSpec()} + dispatch := &rvboxv1.CommandDispatch{IssueUuid: testIssueUUID, CommandRevision: 1, TargetSessionGeneration: 2, IssueTime: fixedTimestamp(), Spec: validCommandSpec(), ImmutableRequestSha256: make([]byte, sha256.Size)} envelope := &rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 2, Payload: &rvboxv1.AgentEnvelope_CommandDispatch{CommandDispatch: dispatch}} if err := ValidateEnvelope(envelope, DefaultLimits(), rvboxv1.Platform_PLATFORM_LINUX); err != nil { t.Fatalf("valid dispatch: %v", err) @@ -149,6 +150,7 @@ func TestCommandDispatchUUIDRevisionAndTimestamp_BH_IDEM_01(t *testing.T) { func(value *rvboxv1.CommandDispatch) { value.IssueUuid = "not-a-uuid" }, func(value *rvboxv1.CommandDispatch) { value.CommandRevision = 0 }, func(value *rvboxv1.CommandDispatch) { value.TargetSessionGeneration = 0 }, + func(value *rvboxv1.CommandDispatch) { value.ImmutableRequestSha256 = nil }, func(value *rvboxv1.CommandDispatch) { value.IssueTime = timestamppb.New(time.Date(10000, 1, 1, 0, 0, 0, 0, time.UTC)) }, diff --git a/internal/server/session/agent_server.go b/internal/server/session/agent_server.go index cd915fc..e161814 100644 --- a/internal/server/session/agent_server.go +++ b/internal/server/session/agent_server.go @@ -257,7 +257,7 @@ func (server *AgentServer) enqueueNextDispatch(ctx context.Context, queue *Write } dispatch := &rvboxv1.CommandDispatch{ IssueUuid: candidate.IssueUUID.String(), CommandRevision: candidate.Revision, TargetSessionGeneration: generation, - IssueTime: timestamppb.New(candidate.IssueTime), Spec: spec, + IssueTime: timestamppb.New(candidate.IssueTime), Spec: spec, ImmutableRequestSha256: candidate.ImmutableSHA256[:], } if candidate.QueueExpiryTime != nil { dispatch.QueueExpiryTime = timestamppb.New(*candidate.QueueExpiryTime) diff --git a/internal/server/store/dispatch.go b/internal/server/store/dispatch.go index 0675b37..73c9da1 100644 --- a/internal/server/store/dispatch.go +++ b/internal/server/store/dispatch.go @@ -23,6 +23,7 @@ type DispatchCandidate struct { Revision uint64 IssueTime time.Time QueueExpiryTime *time.Time + ImmutableSHA256 [32]byte ExecutionSpec []byte } @@ -51,13 +52,13 @@ func (store *Store) ClaimNextDispatch(ctx context.Context, clientID string, gene WHERE client_id = ? AND lifecycle = 1 AND queue_expiry_time IS NOT NULL AND queue_expiry_time <= ?`, now.UTC().UnixNano(), clientID, now.UTC().UnixNano()); err != nil { return nil, err } - var encodedIssue, stored []byte + var encodedIssue, stored, immutableHash []byte var revision uint64 var issueTime int64 var expiry sql.NullInt64 var rawBytes uint64 - err = tx.QueryRowContext(ctx, `SELECT issue_uuid, revision, issue_time, queue_expiry_time, execution_spec, execution_spec_raw_bytes -FROM commands WHERE client_id = ? AND lifecycle = 1 ORDER BY issue_time, issue_uuid LIMIT 1`, clientID).Scan(&encodedIssue, &revision, &issueTime, &expiry, &stored, &rawBytes) + err = tx.QueryRowContext(ctx, `SELECT issue_uuid, revision, issue_time, queue_expiry_time, immutable_request_sha256, execution_spec, execution_spec_raw_bytes +FROM commands WHERE client_id = ? AND lifecycle = 1 ORDER BY issue_time, issue_uuid LIMIT 1`, clientID).Scan(&encodedIssue, &revision, &issueTime, &expiry, &immutableHash, &stored, &rawBytes) if errors.Is(err, sql.ErrNoRows) { if err := tx.Commit(); err != nil { return nil, err @@ -67,11 +68,13 @@ FROM commands WHERE client_id = ? AND lifecycle = 1 ORDER BY issue_time, issue_u if err != nil { return nil, err } - if len(encodedIssue) != 16 { + if len(encodedIssue) != 16 || len(immutableHash) != 32 { return nil, ErrInvalidSegmentRecord } var issue domain.UUID copy(issue[:], encodedIssue) + var immutable [32]byte + copy(immutable[:], immutableHash) spec, err := decompressCommandSpec(stored, rawBytes) if err != nil { return nil, err @@ -90,7 +93,7 @@ FROM commands WHERE client_id = ? AND lifecycle = 1 ORDER BY issue_time, issue_u if err := tx.Commit(); err != nil { return nil, err } - candidate := &DispatchCandidate{IssueUUID: issue, Revision: revision, IssueTime: time.Unix(0, issueTime).UTC(), ExecutionSpec: spec} + candidate := &DispatchCandidate{IssueUUID: issue, Revision: revision, IssueTime: time.Unix(0, issueTime).UTC(), ImmutableSHA256: immutable, ExecutionSpec: spec} if expiry.Valid { value := time.Unix(0, expiry.Int64).UTC() candidate.QueueExpiryTime = &value diff --git a/protos/rvbox/v1/agent.proto b/protos/rvbox/v1/agent.proto index bd3d38b..acc8954 100644 --- a/protos/rvbox/v1/agent.proto +++ b/protos/rvbox/v1/agent.proto @@ -63,6 +63,8 @@ message CommandDispatch { google.protobuf.Timestamp issue_time = 4; ExecutionSpec spec = 5; google.protobuf.Timestamp queue_expiry_time = 6; + // Required immutable digest for client-side UUID/hash idempotency checks. + bytes immutable_request_sha256 = 7; } message CommandAccepted { diff --git a/test/integration/clientagent/clientagent_integration_test.go b/test/integration/clientagent/clientagent_integration_test.go index 25dd13b..52f8413 100644 --- a/test/integration/clientagent/clientagent_integration_test.go +++ b/test/integration/clientagent/clientagent_integration_test.go @@ -106,7 +106,7 @@ func TestWebSocketDispatchAfterReconciliation_HP_DISPATCH_03(t *testing.T) { t.Fatalf("dispatch envelope = %#v, %v", envelope, err) } dispatch := envelope.GetCommandDispatch() - if dispatch == nil || dispatch.GetIssueUuid() != issue.String() || dispatch.GetCommandRevision() != 1 || dispatch.GetTargetSessionGeneration() != accepted.Generation || !proto.Equal(dispatch.GetSpec(), spec) { + if dispatch == nil || dispatch.GetIssueUuid() != issue.String() || dispatch.GetCommandRevision() != 1 || dispatch.GetTargetSessionGeneration() != accepted.Generation || !proto.Equal(dispatch.GetSpec(), spec) || string(dispatch.GetImmutableRequestSha256()) != string(requestHash[:]) { t.Fatalf("dispatch = %#v", dispatch) } }