feat: durably admit client dispatches

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)
+}

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