From ac18ec49e7ef937c42788b7586235a6b274df0e0 Mon Sep 17 00:00:00 2001 From: cabbage Date: Sun, 6 Sep 2026 07:36:45 +0000 Subject: [PATCH] feat: durably admit client dispatches --- internal/client/agent/dispatch.go | 34 +++++++++++++++++++++++++ internal/client/agent/handshake_test.go | 19 ++++++++++++++ 2 files changed, 53 insertions(+) create mode 100644 internal/client/agent/dispatch.go diff --git a/internal/client/agent/dispatch.go b/internal/client/agent/dispatch.go new file mode 100644 index 0000000..44f7a03 --- /dev/null +++ b/internal/client/agent/dispatch.go @@ -0,0 +1,34 @@ +package agent + +import ( + "context" + "errors" + "time" + + rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" + "github.com/rvbox/rvbox/internal/agentproto" + "github.com/rvbox/rvbox/internal/client/spool" + "github.com/rvbox/rvbox/internal/domain" +) + +// PersistDispatch admits a server dispatch to the durable spool before any +// CommandAccepted reply. UUID/hash equality makes a retransmission harmless; +// a different immutable request for the same UUID is rejected by the spool. +func PersistDispatch(ctx context.Context, store *spool.Store, session Session, dispatch *rvboxv1.CommandDispatch, now time.Time, limits agentproto.Limits) (spool.Acceptance, error) { + if store == nil || dispatch == nil || session.Generation == 0 || dispatch.GetTargetSessionGeneration() != session.Generation || now.IsZero() { + return spool.Acceptance{}, ErrProtocolHandshake + } + if err := agentproto.ValidateExecutionSpec(dispatch.GetSpec(), limits, rvboxv1.Platform_PLATFORM_WINDOWS); err != nil { + return spool.Acceptance{}, err + } + issue, err := domain.ParseUUIDv7(dispatch.GetIssueUuid()) + if err != nil || dispatch.GetCommandRevision() == 0 || len(dispatch.GetImmutableRequestSha256()) != 32 { + return spool.Acceptance{}, ErrProtocolHandshake + } + var immutable [32]byte + copy(immutable[:], dispatch.GetImmutableRequestSha256()) + if immutable == [32]byte{} { + return spool.Acceptance{}, errors.New("empty dispatch immutable hash") + } + return store.AcceptCommand(ctx, spool.Command{IssueUUID: issue, ImmutableSHA256: immutable, Revision: dispatch.GetCommandRevision(), Phase: uint32(rvboxv1.CommandLifecycle_COMMAND_ACCEPTED)}, now) +} diff --git a/internal/client/agent/handshake_test.go b/internal/client/agent/handshake_test.go index 6052c2c..7eb6d1a 100644 --- a/internal/client/agent/handshake_test.go +++ b/internal/client/agent/handshake_test.go @@ -3,11 +3,13 @@ package agent import ( "context" "errors" + "path/filepath" "testing" "time" rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1" "github.com/rvbox/rvbox/internal/agentproto" + "github.com/rvbox/rvbox/internal/client/spool" "github.com/rvbox/rvbox/internal/domain" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" @@ -82,6 +84,23 @@ func TestApplyReconcileResult_HP_SES_12(t *testing.T) { } } +func TestPersistDispatchUsesImmutableHash_HP_DISPATCH_04(t *testing.T) { + store, err := spool.Open(context.Background(), spool.Options{DataDir: filepath.Join(t.TempDir(), "spool"), BusyTimeout: time.Second}) + if err != nil { + t.Fatal(err) + } + defer store.Close() + dispatch := &rvboxv1.CommandDispatch{IssueUuid: "019c46f1-1d02-7000-8000-000000000064", CommandRevision: 1, TargetSessionGeneration: 7, IssueTime: timestamppb.Now(), ImmutableRequestSha256: []byte("12345678901234567890123456789012"), Spec: &rvboxv1.ExecutionSpec{ShellType: rvboxv1.ShellType_SHELL_POWERSHELL, Source: &rvboxv1.ExecutionSpec_CommandText{CommandText: "Write-Output ok"}}} + accepted, err := PersistDispatch(context.Background(), store, Session{Generation: 7}, dispatch, time.Now(), agentproto.DefaultLimits()) + if err != nil || accepted.Duplicate { + t.Fatalf("PersistDispatch = %#v, %v", accepted, err) + } + duplicate, err := PersistDispatch(context.Background(), store, Session{Generation: 7}, dispatch, time.Now(), agentproto.DefaultLimits()) + if err != nil || !duplicate.Duplicate { + t.Fatalf("duplicate PersistDispatch = %#v, %v", duplicate, err) + } +} + type fakeTransport struct { written []byte read []byte