From ca85b65557ae4f52ebe8a6c7fd5134a2d12b8cf6 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 27 Jul 2026 19:27:43 +0400 Subject: [PATCH] fix(store): unique event IDs on lease/expiry, honest duplicate-ingest signal (S5, S6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit S5: Store.Lease and Store.ExpireLeases both set Event.ID to the task id, so every TaskLeased/TaskReleased event for a given task collided on ID across every lease of that task — unsound for ApplyAdvisory or any future ID-based lookup. Both now call domain.NewID(). S6: Append's TaskCreated dedup path returned nil (success) without appending anything. main.go's handler then did `s.Events(0)[len(s.Events(0))-1]` and returned that — an unrelated event — with 201 Created, and every other Append caller (Gitea poll/webhook, JSONL ingest) had no way to distinguish "duplicate, as expected" from "genuinely appended". Add domain.ErrDuplicate, returned instead of nil on a duplicate (source, external_id). Add Store.TaskBySource to resolve the already-ingested task by that same dedup key. Update every caller: - main.go's POST /v1/tasks now returns 200 with the existing task on ErrDuplicate instead of fabricating a 201 with the wrong event. - provider.Gitea.Poll/IngestWebhook and provider.JSONL.Ingest treat ErrDuplicate as expected (already-seen issue/line), not a failure — without this, Gitea polling would have errored out of its loop on the first already-ingested issue in every batch, since Poll previously relied on the old nil-on-dup behavior to keep scanning. TestLeaseAndExpireEventIDsAreUnique and TestTaskBySourceResolvesDuplicate cover the store-level fixes; TestAppendReplayAndDeduplicate updated for the new error signal. AUDIT.md S5, S6. --- cmd/orchestra/main.go | 14 +++++++- internal/domain/domain.go | 5 +++ internal/provider/provider.go | 14 +++++--- internal/store/store.go | 19 +++++++++-- internal/store/store_test.go | 63 +++++++++++++++++++++++++++++++++-- 5 files changed, 105 insertions(+), 10 deletions(-) diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index d9aeddb..bb42c63 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "io" "log" "net" @@ -168,6 +169,18 @@ func main() { b, _ := json.Marshal(p) e := domain.Event{ID: id(), Type: "TaskCreated", TaskID: id(), Version: 1, Payload: b, Surface: string(surface(r))} if err := s.Append(e); err != nil { + if errors.Is(err, domain.ErrDuplicate) { + // (source, external_id) was already ingested. Nothing was + // appended; report the existing task idempotently rather + // than fabricating a 201 with an unrelated event (S6). + source, _ := p["source"].(string) + externalID, _ := p["external_id"].(string) + if t, ok := s.TaskBySource(source, externalID); ok { + w.WriteHeader(200) + json.NewEncoder(w).Encode(t) + return + } + } http.Error(w, err.Error(), 400) return } @@ -176,7 +189,6 @@ func main() { log.Printf("route task: %v", err) } } - e = s.Events(0)[len(s.Events(0))-1] w.WriteHeader(201) json.NewEncoder(w).Encode(e) }) diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 9e19c3d..74b6251 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -17,6 +17,11 @@ var ErrConflict = errors.New("task version conflict") var ErrNotFound = errors.New("task not found") var ErrInvalid = errors.New("invalid event") +// ErrDuplicate is returned by Store.Append for a TaskCreated event whose +// (source, external_id) pair was already ingested. The caller already has a +// task for this content; nothing was appended. +var ErrDuplicate = errors.New("duplicate task ingestion") + // CurrentEventSchema is 2: schema 2 requires every event to declare its // authorizing Surface (see ValidateEvent), enforced at the store append // boundary. Schema 1 events already on disk replay unchanged — tolerant diff --git a/internal/provider/provider.go b/internal/provider/provider.go index dc5ec56..c93282b 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -137,10 +137,13 @@ func (j JSONL) Ingest(r io.Reader, sink Sink) (int, error) { return count, fmt.Errorf("line %d: %w", line, err) } b, _ := json.Marshal(p) - if err := sink.Append(domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: domain.NewID(), Version: 1, Payload: b, Surface: string(authz.System)}); err != nil { + err := sink.Append(domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: domain.NewID(), Version: 1, Payload: b, Surface: string(authz.System)}) + if err != nil && !errors.Is(err, domain.ErrDuplicate) { return count, fmt.Errorf("line %d: %w", line, err) } - count++ + if err == nil { + count++ + } } return count, sc.Err() } @@ -321,7 +324,10 @@ func (g Gitea) IngestWebhook(body []byte, signature string, sink Sink) error { project = h.Repository.FullName } } - return sink.Append(g.event(h.Issue, g.sourceName(), project)) + if err := sink.Append(g.event(h.Issue, g.sourceName(), project)); err != nil && !errors.Is(err, domain.ErrDuplicate) { + return err + } + return nil } func validSignature(body []byte, got, secret string) bool { if secret == "" || got == "" { @@ -379,7 +385,7 @@ func (g Gitea) Poll(ctx context.Context, sink Sink) (int, error) { project = g.Repo } for _, i := range issues { - if err := sink.Append(g.event(i, g.sourceName(), project)); err != nil { + if err := sink.Append(g.event(i, g.sourceName(), project)); err != nil && !errors.Is(err, domain.ErrDuplicate) { return 0, err } } diff --git a/internal/store/store.go b/internal/store/store.go index cf3714c..167c26d 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -189,7 +189,7 @@ func (s *Store) Append(e domain.Event) error { return err } if id := s.external[p["source"].(string)+"\x00"+p["external_id"].(string)]; id != "" { - return nil + return domain.ErrDuplicate } } t, taskExists := s.tasks[e.TaskID] @@ -326,6 +326,19 @@ func (s *Store) Task(id string) (domain.Task, bool) { return t, ok } +// TaskBySource resolves the task ingested for a given (source, external_id) +// pair — the dedup key Append.ErrDuplicate rejects re-ingestion against. +func (s *Store) TaskBySource(source, externalID string) (domain.Task, bool) { + s.mu.Lock() + defer s.mu.Unlock() + id := s.external[source+"\x00"+externalID] + if id == "" { + return domain.Task{}, false + } + t, ok := s.tasks[id] + return t, ok +} + func (s *Store) Lease(id, harness string, ttl time.Duration) (domain.Event, error) { if ttl <= 0 { return domain.Event{}, fmt.Errorf("%w: ttl must be positive", domain.ErrInvalid) @@ -338,7 +351,7 @@ func (s *Store) Lease(id, harness string, ttl time.Duration) (domain.Event, erro return domain.Event{}, domain.ErrConflict } p, _ := json.Marshal(map[string]any{"harness_id": harness, "ttl": ttl.Seconds(), "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": t.Version}) - e := domain.Event{ID: id, Type: "TaskLeased", TaskID: id, Version: t.Version + 1, Payload: p, Surface: string(authz.System)} + e := domain.Event{ID: domain.NewID(), Type: "TaskLeased", TaskID: id, Version: t.Version + 1, Payload: p, Surface: string(authz.System)} return e, s.Append(e) } @@ -347,7 +360,7 @@ func (s *Store) ExpireLeases(now time.Time) ([]domain.Event, error) { for _, t := range s.Tasks() { if t.State == domain.StateLeased && t.Lease != nil && !t.Lease.Until.After(now) { p, _ := json.Marshal(map[string]any{"reason": "lease_expired", "harness_id": t.Lease.HarnessID}) - e := domain.Event{ID: t.ID, Type: "TaskReleased", TaskID: t.ID, Version: t.Version + 1, Payload: p, Surface: string(authz.System)} + e := domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: t.ID, Version: t.Version + 1, Payload: p, Surface: string(authz.System)} if err := s.Append(e); err != nil { return out, err } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 2d26598..605afac 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -2,9 +2,11 @@ package store import ( "encoding/json" + "errors" "os" "path/filepath" "testing" + "time" "orchestra/internal/authz" "orchestra/internal/domain" @@ -24,8 +26,8 @@ func TestAppendReplayAndDeduplicate(t *testing.T) { if err := s.Append(created("e1")); err != nil { t.Fatal(err) } - if err := s.Append(created("e2")); err != nil { - t.Fatal(err) + if err := s.Append(created("e2")); !errors.Is(err, domain.ErrDuplicate) { + t.Fatalf("expected ErrDuplicate, got %v", err) } if got := len(s.Events(0)); got != 1 { t.Fatalf("duplicate ingest appended %d events", got) @@ -53,6 +55,63 @@ func TestAppendReplayAndDeduplicate(t *testing.T) { } } +// TestLeaseAndExpireEventIDsAreUnique guards S5: Event.ID was set to the +// task id in both Lease and ExpireLeases, so every lease of the same task +// produced a TaskLeased/TaskReleased event with a colliding ID — unsound +// for ApplyAdvisory or any future ID-based lookup. +func TestLeaseAndExpireEventIDsAreUnique(t *testing.T) { + s, err := Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err := s.Append(created("e1")); err != nil { + t.Fatal(err) + } + task := s.Tasks()[0] + leaseEvt, err := s.Lease(task.ID, "h1", time.Millisecond) + if err != nil { + t.Fatal(err) + } + if leaseEvt.ID == task.ID || leaseEvt.ID == "" { + t.Fatalf("lease event ID %q collides with task ID %q", leaseEvt.ID, task.ID) + } + time.Sleep(2 * time.Millisecond) + expired, err := s.ExpireLeases(time.Now()) + if err != nil { + t.Fatal(err) + } + if len(expired) != 1 { + t.Fatalf("expected 1 expiry, got %d", len(expired)) + } + if expired[0].ID == task.ID || expired[0].ID == leaseEvt.ID || expired[0].ID == "" { + t.Fatalf("expiry event ID %q collides", expired[0].ID) + } +} + +// TestTaskBySourceResolvesDuplicate covers the S6 fix: a caller that gets +// ErrDuplicate from Append must be able to look up the already-ingested +// task by its dedup key instead of guessing at "the last event in the log". +func TestTaskBySourceResolvesDuplicate(t *testing.T) { + s, err := Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err := s.Append(created("e1")); err != nil { + t.Fatal(err) + } + want := s.Tasks()[0] + if err := s.Append(created("e2")); !errors.Is(err, domain.ErrDuplicate) { + t.Fatalf("expected ErrDuplicate, got %v", err) + } + got, ok := s.TaskBySource("jsonl", "42") + if !ok || got.ID != want.ID { + t.Fatalf("TaskBySource = %+v, ok=%v, want %+v", got, ok, want) + } + if _, ok := s.TaskBySource("jsonl", "does-not-exist"); ok { + t.Fatal("expected not found") + } +} + func TestArtifactIsContentAddressed(t *testing.T) { s, err := Open(t.TempDir()) if err != nil {