From 076c3de6ddf0dcd649a0de8b3d21491ccd83b656 Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 26 Jul 2026 20:49:30 +0400 Subject: [PATCH] test: update integration lifecycle fixtures --- internal/integration/end_to_end_test.go | 29 +++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/internal/integration/end_to_end_test.go b/internal/integration/end_to_end_test.go index 6b2b854..19d42fc 100644 --- a/internal/integration/end_to_end_test.go +++ b/internal/integration/end_to_end_test.go @@ -34,6 +34,9 @@ func (h *harness) Release(context.Context, herdr.Session) (string, error) { h.mu.Lock() defer h.mu.Unlock() h.releases++ + // A boundary is a one-shot turn transition. Once released, the next + // lease represents a new turn and must not be released again immediately. + h.boundary = false return h.ref, nil } func (h *harness) Kill(context.Context, herdr.Session) error { @@ -114,12 +117,21 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) { go c.Monitor(ctx, .8, time.Millisecond) deadline := time.Now().Add(time.Second) for time.Now().Before(deadline) { - if got, _ := s.Task(task.ID); got.State == domain.StateQueued { + if h.releases == 1 { break } time.Sleep(time.Millisecond) } got, _ := s.Task(task.ID) + if got.State == domain.StateLeased && h.releases == 1 { + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: got.Version + 1, Payload: mustJSON(map[string]string{ + "handoff_ref": h.ref, + "anchor_sha": "0123456789012345678901234567890123456789", + })}); err != nil { + t.Fatal(err) + } + got, _ = s.Task(task.ID) + } if got.State != domain.StateQueued || h.releases != 1 { t.Fatalf("rotation state=%s releases=%d", got.State, h.releases) } @@ -134,7 +146,10 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) { if err != nil { t.Fatal(err) } - if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: got.Version + 1, Payload: mustJSON(map[string]string{"report_ref": ref})}); err != nil { + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: got.Version + 1, Payload: mustJSON(map[string]any{ + "report_ref": ref, + "receipt": map[string]any{"harness_id": "h1", "consumed": 1}, + })}); err != nil { t.Fatal(err) } got, _ = s.Task(task.ID) @@ -159,7 +174,10 @@ func TestRestartReplayAndReconcileKillsOrphan(t *testing.T) { } // A fresh coordinator sees the durable session, then drops it once the lease is gone. ref, _ := s.PutArtifact([]byte("handoff")) - if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]string{"handoff_ref": ref})}); err != nil { + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]string{ + "handoff_ref": ref, + "anchor_sha": "0123456789012345678901234567890123456789", + })}); err != nil { t.Fatal(err) } s2, err := store.Open(dir) @@ -200,7 +218,10 @@ func TestProviderRetryReflectionQuotaAndVersionConflict(t *testing.T) { t.Fatal(err) } reflector := &fakeReflector{} - if err := (provider.ReflectingSink{Sink: s, Tasks: s, Reflector: reflector}).Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]string{"report_ref": ref})}); err != nil { + if err := (provider.ReflectingSink{Sink: s, Tasks: s, Reflector: reflector}).Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]any{ + "report_ref": ref, + "receipt": map[string]any{"harness_id": "h1", "consumed": 1}, + })}); err != nil { t.Fatal(err) } if reflector.events != 1 {