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