From 0d67af9976b3661547c2cd193211da5d2e1c4956 Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 27 Aug 2026 00:28:57 +0400 Subject: [PATCH] Admit a first lease when a quota limit has no receipt history Burn-in run 2 ingested its task and then sat queued forever. Every herdr in the live config declares quota_limit_5h and quota_limit_weekly, the event log holds zero QuotaReported events, and QuotaSince reported an empty window as unknown. QuotaAvailability fails closed on unknown, so no harness could ever be leased, and the only producer of a receipt is a completed lease. The event log is Orchestra's whole accounting source, so a window holding no receipts is observable zero consumption. QuotaSince now reports known for an empty window and for a harness that has never reported. A receipt that declares its own consumption unknown still fails closed. The refusal also lied about its cause. federatedAvailability collapsed a base gate refusal into the federation health string, so router health said "stale heartbeat or unhealthy local backend" while the heartbeat was one second old. Availability gates now name themselves through an optional ReasonedAvailability contract: quota refusals say whether usage is unknown or the window is exhausted and by how much, and worker refusals distinguish an unregistered worker, a never-probed backend, a stale heartbeat, a stale health check, and an unreachable backend. Co-Authored-By: Claude Opus 5 --- cmd/orchestra/main.go | 28 ++++++-- internal/federation/federation.go | 25 +++++-- internal/federation/federation_test.go | 37 ++++++++++ internal/router/router.go | 69 ++++++++++++++++-- internal/router/router_test.go | 99 +++++++++++++++++++++++++- internal/store/store.go | 12 +++- internal/store/store_test.go | 26 +++++++ 7 files changed, 277 insertions(+), 19 deletions(-) diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 23bd312..b0db0c3 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -102,13 +102,33 @@ func coordinatorOwnsHerdr(h registry.Herdr, localMachine string) bool { } func (a federatedAvailability) Available(h registry.Herdr) bool { - if a.base != nil && !a.base.Available(h) { - return false + return a.Unavailable(h) == "" +} + +// Unavailable keeps the base gate's own reason instead of restating every +// refusal as worker health. A quota refusal reported as a stale heartbeat sent +// burn-in run 2 looking at a worker that was one second fresh. +func (a federatedAvailability) Unavailable(h registry.Herdr) string { + if a.base != nil { + if reason := unavailableReason(a.base, h); reason != "" { + return reason + } } if coordinatorOwnsHerdr(h, a.localMachine) { - return true + return "" } - return a.workers.Available(h.ID) + return a.workers.Unavailable(h.ID) +} + +// unavailableReason mirrors router.unavailableReason for the wrapped base gate. +func unavailableReason(a router.Availability, h registry.Herdr) string { + if ra, ok := a.(router.ReasonedAvailability); ok { + return ra.Unavailable(h) + } + if a.Available(h) { + return "" + } + return "worker unavailable" } func (a federatedAvailability) Supports(h registry.Herdr, project string) bool { diff --git a/internal/federation/federation.go b/internal/federation/federation.go index f8df377..1d5d7a2 100644 --- a/internal/federation/federation.go +++ b/internal/federation/federation.go @@ -413,21 +413,38 @@ func (r *Registry) Heartbeat(id string, health ...WorkerHealth) error { // Available refreshes TTL state and reports whether a registered worker owns // this harness id. Router admission uses it so a reachable TCP bridge alone // can never make an offline worker eligible for a lease. -func (r *Registry) Available(id string) bool { +func (r *Registry) Available(id string) bool { return r.Unavailable(id) == "" } + +// Unavailable refreshes TTL state and returns "" when a registered worker owns +// this harness id and may be leased, otherwise the specific reason. A single +// collapsed reason once reported "stale heartbeat" for a worker whose +// heartbeat was one second old, so each condition names itself. +func (r *Registry) Unavailable(id string) string { r.mu.Lock() defer r.mu.Unlock() r.init() w, ok := r.workers[id] if !ok { - return false + return "worker unavailable: no worker registered for this harness" } // A heartbeat merely proves the worker process can reach the coordinator. // Lease admission additionally requires a fresh probe of the worker's // local execution backend; otherwise a partitioned/down backend still // attracts work. - w.Online = time.Since(w.LastSeen) <= r.TTL && w.Health.HerdrStatus == "reachable" && !w.Health.CheckedAt.IsZero() && time.Since(w.Health.CheckedAt) <= r.TTL + reason := "" + switch { + case time.Since(w.LastSeen) > r.TTL: + reason = "worker unavailable: stale heartbeat" + case w.Health.CheckedAt.IsZero(): + reason = "worker unavailable: backend health never reported" + case time.Since(w.Health.CheckedAt) > r.TTL: + reason = "worker unavailable: stale backend health check" + case w.Health.HerdrStatus != "reachable": + reason = "worker unavailable: backend " + w.Health.HerdrStatus + } + w.Online = reason == "" r.workers[id] = w - return w.Online + return reason } // Supports reports whether an online worker explicitly declared the project. diff --git a/internal/federation/federation_test.go b/internal/federation/federation_test.go index 53f81ce..2c95f67 100644 --- a/internal/federation/federation_test.go +++ b/internal/federation/federation_test.go @@ -268,3 +268,40 @@ func TestResolvedCommandsArePrunedButPendingOnesSurvive(t *testing.T) { t.Fatalf("pending command was pruned: %#v ok=%v", c, ok) } } + +// Each unavailability condition must name itself. One collapsed reason once +// reported a stale heartbeat for a worker whose heartbeat was a second old. +func TestUnavailableNamesTheFailingCondition(t *testing.T) { + r := &Registry{TTL: time.Minute} + if got, want := r.Unavailable("nobody"), "worker unavailable: no worker registered for this harness"; got != want { + t.Fatalf("unregistered reason=%q, want %q", got, want) + } + if err := r.Register(Worker{ID: "w", Token: "t"}, ""); err != nil { + t.Fatal(err) + } + if got, want := r.Unavailable("w"), "worker unavailable: backend health never reported"; got != want { + t.Fatalf("unprobed reason=%q, want %q", got, want) + } + if err := r.Heartbeat("w", WorkerHealth{HerdrStatus: "unreachable", CheckedAt: time.Now().UTC()}); err != nil { + t.Fatal(err) + } + if got, want := r.Unavailable("w"), "worker unavailable: backend unreachable"; got != want { + t.Fatalf("unhealthy-backend reason=%q, want %q", got, want) + } + if err := r.Heartbeat("w", WorkerHealth{HerdrStatus: "reachable", CheckedAt: time.Now().UTC()}); err != nil { + t.Fatal(err) + } + if got := r.Unavailable("w"); got != "" { + t.Fatalf("fresh reachable worker reason=%q, want admitted", got) + } + stale := &Registry{TTL: time.Nanosecond} + if err := stale.Register(Worker{ID: "w", Token: "t"}, ""); err != nil { + t.Fatal(err) + } + if err := stale.Heartbeat("w", WorkerHealth{HerdrStatus: "reachable", CheckedAt: time.Now().UTC()}); err != nil { + t.Fatal(err) + } + if got, want := stale.Unavailable("w"), "worker unavailable: stale heartbeat"; got != want { + t.Fatalf("stale reason=%q, want %q", got, want) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 57d6c78..4407a81 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -3,6 +3,7 @@ package router import ( "encoding/json" + "fmt" "errors" "orchestra/internal/authz" "orchestra/internal/domain" @@ -21,6 +22,15 @@ type Availability interface{ Available(h registry.Herdr) bool } type ProjectAvailability interface { Supports(h registry.Herdr, project string) bool } +// ReasonedAvailability is an optional contract that names why a herdr was +// refused. Without it every refusal collapses into one string, which told an +// operator "stale heartbeat" while the heartbeat was one second old and the +// real refusal came from the quota gate. Found on burn-in run 2, 2026-08-26. +type ReasonedAvailability interface { + // Unavailable returns "" when the herdr may be leased, otherwise the + // specific reason. + Unavailable(h registry.Herdr) string +} type AlwaysAvailable struct{} func (AlwaysAvailable) Available(registry.Herdr) bool { return true } @@ -85,6 +95,53 @@ func (q QuotaAvailability) Available(h registry.Herdr) bool { return true } +func (q QuotaAvailability) Unavailable(h registry.Herdr) string { + if q.Store == nil { + return "quota unavailable: no receipt store" + } + limits, bounded := q.Limits[h.ID] + if !bounded || (limits.FiveHour <= 0 && limits.Weekly <= 0) { + return "" + } + now := time.Now() + if q.Now != nil { + now = q.Now() + } + windows := []struct { + name string + limit float64 + since time.Time + }{ + {"5h", limits.FiveHour, now.Add(-fiveHourWindow)}, + {"weekly", limits.Weekly, now.Add(-weeklyWindow)}, + } + for _, w := range windows { + if w.limit <= 0 { + continue + } + used, known := q.sumSince(h.ID, w.since) + if !known { + return "quota unavailable: " + w.name + " usage unknown" + } + if cap := w.limit * quotaConservativeFraction; used >= cap { + return fmt.Sprintf("quota exhausted: %s window %.0f/%.0f conservative limit (of %.0f)", w.name, used, cap, w.limit) + } + } + return "" +} + +// unavailableReason prefers a specific reason when the availability supports +// one, and never lets a bare Available disagree with it. +func unavailableReason(a Availability, h registry.Herdr) string { + if ra, ok := a.(ReasonedAvailability); ok { + return ra.Unavailable(h) + } + if a.Available(h) { + return "" + } + return "worker unavailable" +} + type RetryPolicy struct { MaxAttempts int Backoff time.Duration @@ -197,7 +254,7 @@ func (r *Router) AssignPending() ([]domain.Event, error) { sort.SliceStable(queued, func(i, j int) bool { return importance(queued[i], now).Before(importance(queued[j], now)) }) health := r.healthSnapshot(now) candidates := make(map[string][]registry.Herdr) - availability := make(map[string]bool) + availability := make(map[string]string) availabilityKnown := make(map[string]bool) projectSupport := make(map[string]bool) projectSupportKnown := make(map[string]bool) @@ -245,13 +302,13 @@ func (r *Router) AssignPending() ([]domain.Event, error) { r.reject(t.ID, h.ID, "capability mismatch") continue } - available, checked := availability[h.ID], availabilityKnown[h.ID] + reason, checked := availability[h.ID], availabilityKnown[h.ID] if !checked { - available = r.Availability.Available(h) - availability[h.ID], availabilityKnown[h.ID] = available, true + reason = unavailableReason(r.Availability, h) + availability[h.ID], availabilityKnown[h.ID] = reason, true } - if !available { - r.reject(t.ID, h.ID, "worker unavailable: stale heartbeat or unhealthy local backend") + if reason != "" { + r.reject(t.ID, h.ID, reason) continue } if occupiedCount(activeLeases[h.ID], h.Concurrency) { diff --git a/internal/router/router_test.go b/internal/router/router_test.go index 01cafd1..7fe2eae 100644 --- a/internal/router/router_test.go +++ b/internal/router/router_test.go @@ -297,9 +297,102 @@ func TestQuotaWindowsAreIndependent(t *testing.T) { if both.Available(registry.Herdr{ID: "h1"}) { t.Fatal("5h window should be exhausted at 90/100 (>=80%) regardless of weekly headroom") } - unknown := QuotaAvailability{Store: s, Limits: map[string]QuotaWindowLimits{"unknown": {FiveHour: 100}}, Now: func() time.Time { return now }} - if unknown.Available(registry.Herdr{ID: "unknown"}) { - t.Fatal("bounded harness without a native usage receipt must fail closed") + // A bounded harness that has never reported is at zero consumption, not + // at unknown consumption: the event log is the whole accounting source. + // Failing closed here deadlocked the first lease, because only a completed + // lease can produce the receipt the gate demanded. + fresh := QuotaAvailability{Store: s, Limits: map[string]QuotaWindowLimits{"unreported": {FiveHour: 100}}, Now: func() time.Time { return now }} + if !fresh.Available(registry.Herdr{ID: "unreported"}) { + t.Fatal("bounded harness with no receipt history must still admit a first lease") + } + // A receipt that declares its own consumption unknown still fails closed. + p, _ := json.Marshal(map[string]any{"harness_id": "opaque", "consumed": 0, "known": false}) + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "QuotaReported", TaskID: "quota", Version: 1, Surface: string(authz.System), Payload: p, At: now}); err != nil { + t.Fatal(err) + } + opaque := QuotaAvailability{Store: s, Limits: map[string]QuotaWindowLimits{"opaque": {FiveHour: 100}}, Now: func() time.Time { return now }} + if opaque.Available(registry.Herdr{ID: "opaque"}) { + t.Fatal("a receipt reporting unknown consumption must fail closed") + } + if got, want := opaque.Unavailable(registry.Herdr{ID: "opaque"}), "quota unavailable: 5h usage unknown"; got != want { + t.Fatalf("unknown-usage reason=%q, want %q", got, want) + } + if got := both.Unavailable(registry.Herdr{ID: "h1"}); !strings.HasPrefix(got, "quota exhausted: 5h window ") { + t.Fatalf("exhausted reason=%q, want a 5h quota-exhausted reason", got) + } +} + +// The router must not restate a quota refusal as worker health. An operator +// reading "stale heartbeat" against a one-second-old heartbeat looks at the +// wrong half of the system. +func TestQuotaRefusalKeepsItsOwnRejectionReason(t *testing.T) { + s, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + r, err := registry.New(registry.Config{ + Projects: []registry.Project{{ID: "p", MachineAffinity: []string{"m"}}}, + Machines: []registry.Machine{{ID: "m", Address: "unused"}}, + Herdrs: []registry.Herdr{{ID: "h1", MachineID: "m", Concurrency: 1}}, + }) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + p, _ := json.Marshal(map[string]any{"harness_id": "h1", "consumed": 90.0}) + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "QuotaReported", TaskID: "quota", Version: 1, Surface: string(authz.System), Payload: p, At: now}); err != nil { + t.Fatal(err) + } + b, _ := json.Marshal(map[string]any{"source": "test", "external_id": "t", "project": "p"}) + if err := s.Append(domain.Event{ID: "t", TaskID: "t", Type: "TaskCreated", Version: 1, Payload: b, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + rt := Router{Store: s, Registry: r, Reachability: reachable{}, + Availability: QuotaAvailability{Store: s, Limits: map[string]QuotaWindowLimits{"h1": {FiveHour: 100}}, Now: func() time.Time { return now }}} + if got, err := rt.AssignPending(); err != nil || len(got) != 0 { + t.Fatalf("assigned %d events, err=%v; want none", len(got), err) + } + found := false + for _, rej := range rt.Rejections() { + if rej.HerdrID != "h1" { + continue + } + found = true + if !strings.HasPrefix(rej.Reason, "quota exhausted: ") { + t.Fatalf("rejection reason=%q, want a quota-exhausted reason", rej.Reason) + } + } + if !found { + t.Fatal("no rejection recorded for h1") + } +} + +// A bounded harness with no receipt history must be leasable, end to end +// through the router. This is the burn-in run 2 blocker: every herdr in the +// live config declares a quota limit, nothing had ever reported a receipt, and +// so no task could ever be assigned autonomously. +func TestFirstLeaseSucceedsWithQuotaConfiguredAndNoReceipts(t *testing.T) { + s, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + r, err := registry.New(registry.Config{ + Projects: []registry.Project{{ID: "p", MachineAffinity: []string{"m"}}}, + Machines: []registry.Machine{{ID: "m", Address: "unused"}}, + Herdrs: []registry.Herdr{{ID: "h1", MachineID: "m", Concurrency: 1}}, + }) + if err != nil { + t.Fatal(err) + } + b, _ := json.Marshal(map[string]any{"source": "test", "external_id": "t", "project": "p"}) + if err := s.Append(domain.Event{ID: "t", TaskID: "t", Type: "TaskCreated", Version: 1, Payload: b, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + rt := Router{Store: s, Registry: r, Reachability: reachable{}, + Availability: QuotaAvailability{Store: s, Limits: map[string]QuotaWindowLimits{"h1": {FiveHour: 50, Weekly: 500}}}} + got, err := rt.AssignPending() + if err != nil || len(got) != 1 { + t.Fatalf("assigned %d events, err=%v; want exactly 1: %+v", len(got), err, rt.Rejections()) } } diff --git a/internal/store/store.go b/internal/store/store.go index a66a51b..c631956 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -910,16 +910,24 @@ func (s *Store) SchedulingSnapshot() SchedulingSnapshot { // QuotaSince answers a rolling-window usage query from the per-harness index // instead of walking events.jsonl. known is false when the interval has no // native receipt or any receipt explicitly reports unknown usage. +// QuotaSince sums the receipts at or after `since`. The event log is +// Orchestra's complete accounting source, so a window holding no receipts is +// observable zero consumption, not missing data: it reports known. `known` is +// false only when a receipt inside the window said its own consumption was +// unknown. Reporting an empty window as unknown deadlocked admission — a +// harness with a configured quota limit and no receipt history could never be +// leased, and the only producer of a receipt is a completed lease. Found on +// burn-in run 2, 2026-08-26. func (s *Store) QuotaSince(harness string, since time.Time) (consumed float64, known bool) { s.mu.Lock() defer s.mu.Unlock() index, ok := s.quota[harness] if !ok { - return 0, false + return 0, true } start := sort.Search(len(index.records), func(i int) bool { return !index.records[i].At.Before(since) }) if start == len(index.records) { - return 0, false + return 0, true } return index.prefix[len(index.records)] - index.prefix[start], index.unknownPrefix[len(index.records)] == index.unknownPrefix[start] } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 9740818..12aca6c 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -637,3 +637,29 @@ func TestReleaseTransactionSurvivesReLeaseUntilMatchingPickup(t *testing.T) { t.Fatalf("pickup not bound to transaction/epoch: %+v", task) } } + +// An empty window is observable zero consumption, not missing data. Reporting +// it as unknown made the router refuse every harness that had a configured +// quota limit and no receipt history, which no first lease could ever produce. +func TestQuotaSinceReportsEmptyWindowAsKnownZero(t *testing.T) { + s, err := Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + if used, known := s.QuotaSince("fresh", now.Add(-fiveHours)); used != 0 || !known { + t.Fatalf("never-reported harness quota=(%v,%v), want (0,true)", used, known) + } + payload, _ := json.Marshal(map[string]any{"harness_id": "h1", "consumed": 7.0, "known": true}) + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "QuotaReported", TaskID: "quota", Version: 1, At: now.Add(-24 * time.Hour), Payload: payload, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + if used, known := s.QuotaSince("h1", now.Add(-fiveHours)); used != 0 || !known { + t.Fatalf("receipts only outside the window quota=(%v,%v), want (0,true)", used, known) + } + if used, known := s.QuotaSince("h1", now.Add(-48*time.Hour)); used != 7 || !known { + t.Fatalf("receipts inside the window quota=(%v,%v), want (7,true)", used, known) + } +} + +const fiveHours = 5 * time.Hour