feat: add revisioned resync reducer

server/domain/sync.go

@@ -0,0 +1,81 @@
+package domain
+
+import "fmt"
+
+var (
+	ErrRevisionGap  = fmt.Errorf("revision gap requires resync")
+	ErrSyncConflict = fmt.Errorf("conflicting revisioned event")
+)
+
+type SyncEvent struct {
+	Kind       ResourceKind
+	ResourceID string
+	Revision   uint64
+	State      State
+}
+
+type ReplicaResource struct {
+	Kind        ResourceKind
+	ResourceID  string
+	State       State
+	Revision    uint64
+	NeedsResync bool
+}
+
+func NewReplicaResource(kind ResourceKind, resourceID string, state State) (*ReplicaResource, error) {
+	if resourceID == "" || !validStateForKind(kind, state) {
+		return nil, fmt.Errorf("invalid replica resource")
+	}
+	return &ReplicaResource{Kind: kind, ResourceID: resourceID, State: state}, nil
+}
+
+// ApplyEvent makes duplicate/out-of-order delivery converge. A gap is not
+// guessed through; callers must fetch the authoritative REST snapshot and use
+// ReplaceSnapshot before resuming the event stream.
+func (r *ReplicaResource) ApplyEvent(event SyncEvent) error {
+	if event.Kind != r.Kind || event.ResourceID != r.ResourceID {
+		return ErrSyncConflict
+	}
+	if r.NeedsResync {
+		return ErrRevisionGap
+	}
+	if event.Revision <= r.Revision {
+		if event.Revision == r.Revision && event.State != r.State {
+			return ErrSyncConflict
+		}
+		return nil
+	}
+	if event.Revision != r.Revision+1 {
+		r.NeedsResync = true
+		return ErrRevisionGap
+	}
+	if !legalTransition(r.Kind, r.State, event.State) {
+		return ErrSyncConflict
+	}
+	r.State, r.Revision = event.State, event.Revision
+	return nil
+}
+
+func (r *ReplicaResource) ReplaceSnapshot(revision uint64, state State) error {
+	if revision < r.Revision || !validStateForKind(r.Kind, state) {
+		return ErrSyncConflict
+	}
+	r.State, r.Revision, r.NeedsResync = state, revision, false
+	return nil
+}
+
+func validStateForKind(kind ResourceKind, state State) bool {
+	switch kind {
+	case ResourceQueueTicket:
+		_, ok := queueTransitions[state]
+		return ok || state == Queued
+	case ResourceProposal:
+		_, ok := proposalTransitions[state]
+		return ok || state == Open
+	case ResourceMatch:
+		_, ok := matchTransitions[state]
+		return ok
+	default:
+		return false
+	}
+}

server/domain/sync_test.go

@@ -0,0 +1,51 @@
+package domain
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestRevisionedReplicaRejectsGapAndConvergesAfterAuthoritativeSnapshot(t *testing.T) {
+	r, err := NewReplicaResource(ResourceQueueTicket, "ticket-1", Queued)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := r.ApplyEvent(SyncEvent{Kind: ResourceQueueTicket, ResourceID: "ticket-1", Revision: 2, State: Accepted}); !errors.Is(err, ErrRevisionGap) || !r.NeedsResync {
+		t.Fatalf("gap = %+v err=%v", r, err)
+	}
+	if err := r.ApplyEvent(SyncEvent{Kind: ResourceQueueTicket, ResourceID: "ticket-1", Revision: 1, State: Proposed}); !errors.Is(err, ErrRevisionGap) {
+		t.Fatalf("event applied while resync required: %v", err)
+	}
+	if err := r.ReplaceSnapshot(2, Accepted); err != nil || r.NeedsResync {
+		t.Fatalf("snapshot = %+v err=%v", r, err)
+	}
+	if err := r.ApplyEvent(SyncEvent{Kind: ResourceQueueTicket, ResourceID: "ticket-1", Revision: 3, State: Allocating}); err != nil || r.Revision != 3 {
+		t.Fatalf("resume = %+v err=%v", r, err)
+	}
+}
+
+func TestReplicaRejectsInvalidInitialState(t *testing.T) {
+	if _, err := NewReplicaResource(ResourceProposal, "proposal-1", Live); err == nil {
+		t.Fatal("invalid proposal state accepted")
+	}
+}
+
+func TestRevisionedReplicaMakesDuplicateAndOutOfOrderEventsIdempotent(t *testing.T) {
+	r, _ := NewReplicaResource(ResourceQueueTicket, "ticket-1", Queued)
+	event := SyncEvent{Kind: ResourceQueueTicket, ResourceID: "ticket-1", Revision: 1, State: Proposed}
+	if err := r.ApplyEvent(event); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.ApplyEvent(event); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.ApplyEvent(SyncEvent{Kind: ResourceQueueTicket, ResourceID: "ticket-1", Revision: 0, State: Queued}); err != nil {
+		t.Fatal(err)
+	}
+	if r.Revision != 1 || r.State != Proposed {
+		t.Fatalf("replay changed state: %+v", r)
+	}
+	if err := r.ApplyEvent(SyncEvent{Kind: ResourceQueueTicket, ResourceID: "ticket-1", Revision: 1, State: Accepted}); !errors.Is(err, ErrSyncConflict) {
+		t.Fatalf("same-revision conflict = %v", err)
+	}
+}