From fbb13c89d2a54031af733961acf8d762005a474e Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 30 Jul 2026 15:36:23 +0400 Subject: [PATCH] perf: index routing snapshots and quota usage --- AUDIT.md | 16 ++- internal/registry/registry.go | 41 +++++--- internal/router/benchmark_test.go | 82 ++++++++++++++++ internal/router/router.go | 146 ++++++++++++++++++++-------- internal/router/router_test.go | 94 ++++++++++++++++++ internal/store/benchmark_test.go | 60 ++++++++++++ internal/store/store.go | 155 ++++++++++++++++++++++++++++-- internal/store/store_test.go | 46 +++++++++ 8 files changed, 577 insertions(+), 63 deletions(-) create mode 100644 internal/router/benchmark_test.go create mode 100644 internal/store/benchmark_test.go diff --git a/AUDIT.md b/AUDIT.md index 25323b8..4e16f3c 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -64,11 +64,17 @@ released agent, or reject a valid completion. ## P2 — performance -- Cache/parallelise health probes; schedule from one task/worker snapshot. - Current routing repeatedly scans tasks and probes candidates per queued task. -- Index active leases and quota windows. Do not scan the whole event log per - availability check or rewrite the full task snapshot after every event. -- Add 1k/10k-task benchmarks with assignment and append p95 budgets. +- **Closed 2026-07-30.** Each scheduling pass takes one atomic task/lease + snapshot, batches cached (TTL) reachability probes concurrently, and + evaluates candidate availability once. It no longer probes candidates or + scans active tasks once per queued task. +- **Closed 2026-07-30.** Active leases and per-harness, time-ordered quota + receipts are projection indexes. Availability uses indexed rolling-window + sums rather than decoding the event log; the task snapshot is a one-time + disposable compatibility cache rather than a full rewrite on every append. +- **Closed 2026-07-30.** `BenchmarkAssignPending{1K,10K}` and + `BenchmarkAppend{1K,10K}` report and enforce p95 budgets, with durable + fsync cost included in their respective paths. ## Delivery order diff --git a/internal/registry/registry.go b/internal/registry/registry.go index 8c36134..7a0e916 100644 --- a/internal/registry/registry.go +++ b/internal/registry/registry.go @@ -227,6 +227,20 @@ func (TCPReachability) Reachable(address string, timeout time.Duration) bool { return true } func (r Registry) Candidates(project string, check Reachability, timeout time.Duration) ([]Herdr, error) { + return r.candidates(project, func(h Herdr) bool { + address := r.Endpoint(h) + return check == nil || check.Reachable(address, timeout) + }) +} + +// CandidatesWithHealth filters candidates using a health snapshot gathered by +// the router. Keeping probe execution outside this method lets a scheduling +// pass probe each herdr once, in parallel, instead of once per queued task. +func (r Registry) CandidatesWithHealth(project string, healthy map[string]bool) ([]Herdr, error) { + return r.candidates(project, func(h Herdr) bool { return healthy[h.ID] }) +} + +func (r Registry) candidates(project string, include func(Herdr) bool) ([]Herdr, error) { p, ok := r.projects[project] if !ok { return nil, ErrUnknownProject @@ -237,20 +251,25 @@ func (r Registry) Candidates(project string, check Reachability, timeout time.Du } out := []Herdr{} for _, h := range r.herdrs { - if !allowed[h.MachineID] { + if !allowed[h.MachineID] || !include(h) { continue } - addr := h.Address - if addr == "" { - addr = r.machines[h.MachineID].Address - if host, _, err := net.SplitHostPort(addr); err == nil { - addr = net.JoinHostPort(host, defaultHerdrPort) - } - } - if check == nil || check.Reachable(addr, timeout) { - out = append(out, h) - } + out = append(out, h) } sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) return out, nil } + +// Endpoint resolves the herdr-specific address or its machine's default +// herdr endpoint. It is exposed so health checks can be batched independently +// from project routing. +func (r Registry) Endpoint(h Herdr) string { + if h.Address != "" { + return h.Address + } + address := r.machines[h.MachineID].Address + if host, _, err := net.SplitHostPort(address); err == nil { + return net.JoinHostPort(host, defaultHerdrPort) + } + return address +} diff --git a/internal/router/benchmark_test.go b/internal/router/benchmark_test.go new file mode 100644 index 0000000..12bf27d --- /dev/null +++ b/internal/router/benchmark_test.go @@ -0,0 +1,82 @@ +package router + +import ( + "encoding/json" + "fmt" + "orchestra/internal/authz" + "orchestra/internal/domain" + "orchestra/internal/registry" + "orchestra/internal/store" + "sort" + "testing" + "time" +) + +// These are p95 budgets for a complete scheduling pass on an otherwise idle +// coordinator. The benchmark deliberately excludes fixture construction and +// disk population, which is measured separately by store's append benchmark. +const ( + // Assignment retains one fsynced TaskLeased event per task; these budgets + // include that durability cost on CI-backed storage. + assignmentP95Budget1K = 6 * time.Second + assignmentP95Budget10K = time.Minute +) + +func BenchmarkAssignPending1K(b *testing.B) { benchmarkAssignPending(b, 1_000, assignmentP95Budget1K) } +func BenchmarkAssignPending10K(b *testing.B) { + benchmarkAssignPending(b, 10_000, assignmentP95Budget10K) +} + +func benchmarkAssignPending(b *testing.B, tasks int, budget time.Duration) { + b.Helper() + samples := make([]time.Duration, 0, b.N) + b.ResetTimer() + for sample := 0; sample < b.N; sample++ { + b.StopTimer() + s, err := store.Open(b.TempDir()) + if err != nil { + b.Fatal(err) + } + for i := 0; i < tasks; i++ { + id := fmt.Sprintf("task-%d", i) + payload, _ := json.Marshal(map[string]any{"source": "benchmark", "external_id": id, "project": "p"}) + if err := s.Append(domain.Event{ID: id, Type: "TaskCreated", TaskID: id, Version: 1, Payload: payload, Surface: string(authz.System)}); err != nil { + b.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: "h", MachineID: "m"}}, // unlimited concurrency + }) + if err != nil { + b.Fatal(err) + } + router := Router{Store: s, Registry: r} + b.StartTimer() + started := time.Now() + assigned, err := router.AssignPending() + samples = append(samples, time.Since(started)) + b.StopTimer() + if err != nil { + b.Fatal(err) + } + if len(assigned) != tasks { + b.Fatalf("assigned %d tasks, want %d", len(assigned), tasks) + } + } + p95 := durationP95(samples) + b.ReportMetric(float64(p95.Microseconds()), "assign_p95_us") + if p95 > budget { + b.Fatalf("assignment p95 %s exceeds budget %s for %d tasks", p95, budget, tasks) + } +} + +func durationP95(samples []time.Duration) time.Duration { + if len(samples) == 0 { + return 0 + } + sorted := append([]time.Duration(nil), samples...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] }) + return sorted[(len(sorted)*95+99)/100-1] +} diff --git a/internal/router/router.go b/internal/router/router.go index 9c95f89..c401488 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -10,6 +10,7 @@ import ( "orchestra/internal/store" "sort" "strings" + "sync" "time" ) @@ -54,26 +55,7 @@ type QuotaAvailability struct { } func (q QuotaAvailability) sumSince(harnessID string, since time.Time) (float64, bool) { - var consumed float64 - known := false - for _, e := range q.Store.Events(0) { - if e.Type != "QuotaReported" || e.At.Before(since) { - continue - } - var p struct { - HarnessID string `json:"harness_id"` - Consumed float64 `json:"consumed"` - Known *bool `json:"known"` - } - if json.Unmarshal(e.Payload, &p) == nil && p.HarnessID == harnessID && p.Consumed >= 0 { - if p.Known != nil && !*p.Known { - return 0, false - } - known = true - consumed += p.Consumed - } - } - return consumed, known + return q.Store.QuotaSince(harnessID, since) } func (q QuotaAvailability) Available(h registry.Herdr) bool { @@ -116,6 +98,18 @@ type Router struct { Retry RetryPolicy Now func() time.Time OnLease func(domain.Event) error + // HealthTTL bounds re-use of a successful or failed reachability probe. + // A scheduling pass always has a coherent health snapshot; this cache also + // prevents bursts of TaskCreated events from repeatedly dialing the same + // herdr between passes. + HealthTTL time.Duration + healthMu sync.Mutex + health map[string]healthProbe +} + +type healthProbe struct { + reachable bool + until time.Time } func (r *Router) init() { @@ -125,6 +119,9 @@ func (r *Router) init() { if r.Now == nil { r.Now = time.Now } + if r.HealthTTL <= 0 { + r.HealthTTL = 5 * time.Second + } } // HandleEvent evaluates the sink after creation and after a lease is freed. @@ -141,13 +138,22 @@ func (r *Router) AssignPending() ([]domain.Event, error) { if r.Store == nil { return nil, errors.New("router: store required") } + now := r.Now() + snapshot := r.Store.SchedulingSnapshot() var queued []domain.Task - for _, t := range r.Store.Tasks() { - if t.State == domain.StateQueued && (t.NextRetryAt.IsZero() || !r.Now().Before(t.NextRetryAt)) { + for _, t := range snapshot.Tasks { + if t.State == domain.StateQueued && (t.NextRetryAt.IsZero() || !now.Before(t.NextRetryAt)) { queued = append(queued, t) } } - sort.SliceStable(queued, func(i, j int) bool { return importance(queued[i], r.Now()).Before(importance(queued[j], r.Now())) }) + 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) + availabilityKnown := make(map[string]bool) + projectSupport := make(map[string]bool) + projectSupportKnown := make(map[string]bool) + activeLeases := snapshot.ActiveLeases var out []domain.Event for _, t := range queued { if r.Retry.MaxAttempts > 0 && t.Attempt >= r.Retry.MaxAttempts { @@ -158,16 +164,34 @@ func (r *Router) AssignPending() ([]domain.Event, error) { out = append(out, e) continue } - cs, err := r.Registry.Candidates(t.Project, r.Reachability, r.Timeout) - if err != nil { - continue + cs, found := candidates[t.Project] + if !found { + var err error + cs, err = r.Registry.CandidatesWithHealth(t.Project, health) + if err != nil { + continue + } + candidates[t.Project] = cs } for _, h := range cs { - projectOK := true - if projects, ok := r.Availability.(ProjectAvailability); ok { - projectOK = projects.Supports(h, t.Project) + projectKey := h.ID + "\x00" + t.Project + projectOK, checked := projectSupport[projectKey], projectSupportKnown[projectKey] + if !checked { + projectOK = true + if projects, ok := r.Availability.(ProjectAvailability); ok { + projectOK = projects.Supports(h, t.Project) + } + projectSupport[projectKey], projectSupportKnown[projectKey] = projectOK, true } - if !projectOK || !matches(t.Capability, h.Capabilities) || !r.Availability.Available(h) || occupied(r.Store, h.ID, h.Concurrency) { + if !projectOK || !matches(t.Capability, h.Capabilities) { + continue + } + available, checked := availability[h.ID], availabilityKnown[h.ID] + if !checked { + available = r.Availability.Available(h) + availability[h.ID], availabilityKnown[h.ID] = available, true + } + if !available || occupiedCount(activeLeases[h.ID], h.Concurrency) { continue } e, err := r.Store.Lease(t.ID, h.ID, 30*time.Minute) @@ -175,6 +199,7 @@ func (r *Router) AssignPending() ([]domain.Event, error) { continue } out = append(out, e) + activeLeases[h.ID]++ if r.OnLease != nil { if err := r.OnLease(e); err != nil { return out, err @@ -198,17 +223,60 @@ func matches(need, have []string) bool { } return true } -func occupied(s *store.Store, id string, limit int) bool { - if limit <= 0 { - return false - } - n := 0 - for _, t := range s.Tasks() { - if (t.State == domain.StateLeased || t.State == domain.StateNeedsAttention) && t.Lease != nil && t.Lease.HarnessID == id { - n++ +func occupiedCount(count, limit int) bool { + return limit > 0 && count >= limit +} + +func (r *Router) healthSnapshot(now time.Time) map[string]bool { + herdrs := r.Registry.Herdrs() + health := make(map[string]bool, len(herdrs)) + if r.Reachability == nil { + for _, h := range herdrs { + health[h.ID] = true } + return health } - return n >= limit + type target struct { + key string + id string + address string + } + var probes []target + r.healthMu.Lock() + if r.health == nil { + r.health = make(map[string]healthProbe) + } + for _, h := range herdrs { + key := h.ID + "\x00" + r.Registry.Endpoint(h) + if cached, ok := r.health[key]; ok && now.Before(cached.until) { + health[h.ID] = cached.reachable + continue + } + probes = append(probes, target{key: key, id: h.ID, address: r.Registry.Endpoint(h)}) + } + r.healthMu.Unlock() + var wg sync.WaitGroup + var mu sync.Mutex + for _, target := range probes { + target := target + wg.Add(1) + go func() { + defer wg.Done() + reachable := r.Reachability.Reachable(target.address, r.Timeout) + mu.Lock() + health[target.id] = reachable + mu.Unlock() + }() + } + wg.Wait() + if len(probes) > 0 { + r.healthMu.Lock() + for _, target := range probes { + r.health[target.key] = healthProbe{reachable: health[target.id], until: now.Add(r.HealthTTL)} + } + r.healthMu.Unlock() + } + return health } func importance(t domain.Task, now time.Time) time.Time { if t.Due != nil { diff --git a/internal/router/router_test.go b/internal/router/router_test.go index f01a6fe..ec4c328 100644 --- a/internal/router/router_test.go +++ b/internal/router/router_test.go @@ -6,6 +6,7 @@ import ( "orchestra/internal/domain" "orchestra/internal/registry" "orchestra/internal/store" + "sync/atomic" "testing" "time" ) @@ -14,6 +15,26 @@ type reachable struct{} func (reachable) Reachable(string, time.Duration) bool { return true } +type countedReachability struct { + calls atomic.Int32 + delay time.Duration +} + +func (r *countedReachability) Reachable(string, time.Duration) bool { + r.calls.Add(1) + if r.delay > 0 { + time.Sleep(r.delay) + } + return true +} + +type countedAvailability struct{ calls atomic.Int32 } + +func (a *countedAvailability) Available(registry.Herdr) bool { + a.calls.Add(1) + return true +} + type projectAvailability struct{ projects map[string]bool } func (p projectAvailability) Available(registry.Herdr) bool { return true } @@ -63,6 +84,79 @@ func TestAssignsByAffinityCapabilityAndConcurrency(t *testing.T) { } } +func TestAssignPendingSnapshotsAndCachesHealth(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", Address: "h1:1"}, + {ID: "h2", MachineID: "m", Address: "h2:1"}, + {ID: "h3", MachineID: "m", Address: "h3:1"}, + }, + }) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 10; i++ { + id := domain.NewID() + payload, _ := json.Marshal(map[string]any{"source": "test", "external_id": id, "project": "p"}) + if err := s.Append(domain.Event{ID: id, TaskID: id, Type: "TaskCreated", Version: 1, Payload: payload, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + } + reach := &countedReachability{} + availability := &countedAvailability{} + rt := Router{Store: s, Registry: r, Reachability: reach, Availability: availability, HealthTTL: time.Minute} + if got, err := rt.AssignPending(); err != nil || len(got) != 10 { + t.Fatalf("assigned=%d, err=%v", len(got), err) + } + if got := reach.calls.Load(); got != 3 { + t.Fatalf("health probes=%d, want one per herdr rather than one per task", got) + } + if got := availability.calls.Load(); got != 1 { + t.Fatalf("availability checks=%d, want one for selected harness", got) + } + // The next pass has no eligible work, but still obtains a health snapshot. + // It must use the TTL cache rather than re-dial all three herdrs. + if _, err := rt.AssignPending(); err != nil { + t.Fatal(err) + } + if got := reach.calls.Load(); got != 3 { + t.Fatalf("health cache missed: probes=%d, want 3", got) + } +} + +func TestHealthProbesRunInParallel(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", Address: "h1:1"}, {ID: "h2", MachineID: "m", Address: "h2:1"}, {ID: "h3", MachineID: "m", Address: "h3:1"}}, + }) + if err != nil { + t.Fatal(err) + } + payload, _ := json.Marshal(map[string]any{"source": "test", "external_id": "parallel", "project": "p"}) + if err := s.Append(domain.Event{ID: "parallel", TaskID: "parallel", Type: "TaskCreated", Version: 1, Payload: payload, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + reach := &countedReachability{delay: 100 * time.Millisecond} + started := time.Now() + if _, err := (&Router{Store: s, Registry: r, Reachability: reach}).AssignPending(); err != nil { + t.Fatal(err) + } + if elapsed := time.Since(started); elapsed > 250*time.Millisecond { + t.Fatalf("health probes took %s; expected parallel probes, not ~300ms serial", elapsed) + } +} + func TestAssignPendingRequiresWorkerProjectSupport(t *testing.T) { s, err := store.Open(t.TempDir()) if err != nil { diff --git a/internal/store/benchmark_test.go b/internal/store/benchmark_test.go new file mode 100644 index 0000000..59888a6 --- /dev/null +++ b/internal/store/benchmark_test.go @@ -0,0 +1,60 @@ +package store + +import ( + "encoding/json" + "fmt" + "orchestra/internal/authz" + "orchestra/internal/domain" + "sort" + "testing" + "time" +) + +// Append includes the fsynced event-log write and sparse projection-cache +// checkpoint. The budgets catch accidental return to an O(tasks) snapshot +// rewrite on every event while remaining practical on CI-backed storage. +const ( + appendP95Budget1K = 10 * time.Second + appendP95Budget10K = 90 * time.Second +) + +func BenchmarkAppend1K(b *testing.B) { benchmarkAppend(b, 1_000, appendP95Budget1K) } +func BenchmarkAppend10K(b *testing.B) { benchmarkAppend(b, 10_000, appendP95Budget10K) } + +func benchmarkAppend(b *testing.B, events int, budget time.Duration) { + b.Helper() + samples := make([]time.Duration, 0, b.N) + b.ResetTimer() + for sample := 0; sample < b.N; sample++ { + b.StopTimer() + s, err := Open(b.TempDir()) + if err != nil { + b.Fatal(err) + } + b.StartTimer() + started := time.Now() + for i := 0; i < events; i++ { + id := fmt.Sprintf("task-%d", i) + payload, _ := json.Marshal(map[string]any{"source": "benchmark", "external_id": id, "project": "p"}) + if err := s.Append(domain.Event{ID: id, Type: "TaskCreated", TaskID: id, Version: 1, Payload: payload, Surface: string(authz.System)}); err != nil { + b.Fatal(err) + } + } + samples = append(samples, time.Since(started)) + b.StopTimer() + } + p95 := appendDurationP95(samples) + b.ReportMetric(float64(p95.Microseconds()), "append_p95_us") + if p95 > budget { + b.Fatalf("append p95 %s exceeds budget %s for %d events", p95, budget, events) + } +} + +func appendDurationP95(samples []time.Duration) time.Duration { + if len(samples) == 0 { + return 0 + } + sorted := append([]time.Duration(nil), samples...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] }) + return sorted[(len(sorted)*95+99)/100-1] +} diff --git a/internal/store/store.go b/internal/store/store.go index 41fcdad..7347fcb 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -9,10 +9,26 @@ import ( "orchestra/internal/domain" "os" "path/filepath" + "sort" "sync" "time" ) +// QuotaUsage is an indexed native-usage receipt. It is deliberately kept +// separate from the event log: event history remains authoritative, while +// routing must not decode every historical event for each availability check. +type QuotaUsage struct { + At time.Time + Consumed float64 + Known bool +} + +type quotaIndex struct { + records []QuotaUsage // ordered by At + prefix []float64 + unknownPrefix []int +} + type Store struct { mu sync.Mutex path string @@ -20,15 +36,19 @@ type Store struct { events []domain.Event tasks map[string]domain.Task external map[string]string - snapshot string - seq uint64 + // activeLeases is the routing occupancy index. Needs-attention retains a + // fenced owner, so it counts as active until that lease is released. + activeLeases map[string]map[string]struct{} + quota map[string]quotaIndex + snapshot string + seq uint64 } func Open(dir string) (*Store, error) { if err := os.MkdirAll(dir, 0755); err != nil { return nil, err } - s := &Store{path: filepath.Join(dir, "events.jsonl"), cas: filepath.Join(dir, "cas"), snapshot: filepath.Join(dir, "snapshot.json"), tasks: map[string]domain.Task{}, external: map[string]string{}} + s := &Store{path: filepath.Join(dir, "events.jsonl"), cas: filepath.Join(dir, "cas"), snapshot: filepath.Join(dir, "snapshot.json"), tasks: map[string]domain.Task{}, external: map[string]string{}, activeLeases: map[string]map[string]struct{}{}, quota: map[string]quotaIndex{}} if err := os.MkdirAll(s.cas, 0755); err != nil { return nil, err } @@ -58,7 +78,8 @@ func Open(dir string) (*Store, error) { if t, ok := s.tasks[e.TaskID]; ok && e.Version != t.Version+1 { return nil, domain.ErrConflict } - if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" { + global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" + if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" && !global { return nil, domain.ErrNotFound } s.events = append(s.events, e) @@ -79,7 +100,17 @@ func (s *Store) apply(e domain.Event) error { return err } t := s.tasks[e.TaskID] - if e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" { + if e.Type == "QuotaReported" { + known := true + if v, ok := p["known"].(bool); ok { + known = v + } + harness, _ := p["harness_id"].(string) + consumed, _ := p["consumed"].(float64) + s.addQuotaUsage(harness, QuotaUsage{At: e.At, Consumed: consumed, Known: known}) + return nil + } + if e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" { return nil } switch e.Type { @@ -266,10 +297,72 @@ func (s *Store) apply(e domain.Event) error { } } t.Version = e.Version - s.tasks[e.TaskID] = t + s.replaceTask(e.TaskID, t) return nil } +func activeLease(t domain.Task) (string, bool) { + if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil { + return "", false + } + return t.Lease.HarnessID, t.Lease.HarnessID != "" +} + +func (s *Store) replaceTask(id string, next domain.Task) { + if previous, ok := s.tasks[id]; ok { + if harness, active := activeLease(previous); active { + delete(s.activeLeases[harness], id) + if len(s.activeLeases[harness]) == 0 { + delete(s.activeLeases, harness) + } + } + } + if harness, active := activeLease(next); active { + if s.activeLeases[harness] == nil { + s.activeLeases[harness] = map[string]struct{}{} + } + s.activeLeases[harness][id] = struct{}{} + } + s.tasks[id] = next +} + +func (s *Store) addQuotaUsage(harness string, usage QuotaUsage) { + index := s.quota[harness] + // Receipts normally arrive in timestamp order, making index maintenance + // O(1). Keep the out-of-order path correct for replay and delayed worker + // reports without imposing its rebuild cost on the common append path. + at := sort.Search(len(index.records), func(i int) bool { return usage.At.Before(index.records[i].At) }) + if at == len(index.records) { + if len(index.prefix) == 0 { + index.prefix = []float64{0} + index.unknownPrefix = []int{0} + } + prefix := index.prefix[len(index.prefix)-1] + unknown := index.unknownPrefix[len(index.unknownPrefix)-1] + index.records = append(index.records, usage) + index.prefix = append(index.prefix, prefix+usage.Consumed) + if !usage.Known { + unknown++ + } + index.unknownPrefix = append(index.unknownPrefix, unknown) + s.quota[harness] = index + return + } + index.records = append(index.records, QuotaUsage{}) + copy(index.records[at+1:], index.records[at:]) + index.records[at] = usage + index.prefix = make([]float64, len(index.records)+1) + index.unknownPrefix = make([]int, len(index.records)+1) + for i, receipt := range index.records { + index.prefix[i+1] = index.prefix[i] + receipt.Consumed + index.unknownPrefix[i+1] = index.unknownPrefix[i] + if !receipt.Known { + index.unknownPrefix[i+1]++ + } + } + s.quota[harness] = index +} + func retryBackoff(attempt int) time.Duration { if attempt < 1 { attempt = 1 @@ -392,8 +485,13 @@ func (s *Store) Append(e domain.Event) error { s.events = append(s.events, e) s.seq = e.Seq // Snapshot failure does not roll back a committed event. Open always - // rebuilds from the log, so leaving a stale cache is safe. - _ = s.writeSnapshot() + // rebuilds from the log, so leaving a stale cache is safe. Keep only the + // initial compatibility cache: repeatedly serializing the full projection + // turns an otherwise O(1) append into O(tasks) work and is never used for + // recovery. + if s.seq == 1 { + _ = s.writeSnapshot() + } return nil } @@ -491,6 +589,47 @@ func (s *Store) Tasks() []domain.Task { } return out } + +// SchedulingSnapshot captures the task projection and lease occupancy under +// one lock. A routing pass uses this immutable view, then updates its local +// occupancy as it issues leases; it never repeatedly scan-locks the store. +type SchedulingSnapshot struct { + Tasks []domain.Task + ActiveLeases map[string]int +} + +func (s *Store) SchedulingSnapshot() SchedulingSnapshot { + s.mu.Lock() + defer s.mu.Unlock() + snapshot := SchedulingSnapshot{ + Tasks: make([]domain.Task, 0, len(s.tasks)), + ActiveLeases: make(map[string]int, len(s.activeLeases)), + } + for _, task := range s.tasks { + snapshot.Tasks = append(snapshot.Tasks, task) + } + for harness, leases := range s.activeLeases { + snapshot.ActiveLeases[harness] = len(leases) + } + return snapshot +} + +// 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. +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 + } + 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 index.prefix[len(index.records)] - index.prefix[start], index.unknownPrefix[len(index.records)] == index.unknownPrefix[start] +} func (s *Store) Events(since uint64) []domain.Event { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/store/store_test.go b/internal/store/store_test.go index fe159fa..9740818 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -212,6 +212,52 @@ func TestOpenRebuildsOnlyFromLogAndIgnoresCorruptSnapshot(t *testing.T) { } } +func TestSchedulingAndQuotaIndexesReplayFromEvents(t *testing.T) { + dir := t.TempDir() + s, err := Open(dir) + if err != nil { + t.Fatal(err) + } + if err := s.Append(created("indexed")); err != nil { + t.Fatal(err) + } + if _, err := s.Lease("task-1", "h1", time.Hour); err != nil { + t.Fatal(err) + } + snapshot := s.SchedulingSnapshot() + if got := snapshot.ActiveLeases["h1"]; got != 1 { + t.Fatalf("active lease index=%d, want 1", got) + } + now := time.Now().UTC() + report := func(at time.Time, consumed float64, known bool) { + payload, _ := json.Marshal(map[string]any{"harness_id": "h1", "consumed": consumed, "known": known}) + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "QuotaReported", TaskID: "quota", Version: 1, At: at, Payload: payload, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + } + // Deliberately append out of timestamp order: the index must answer the + // rolling query by receipt time, not event sequence. + report(now, 4, true) + report(now.Add(-2*time.Hour), 3, true) + report(now.Add(-time.Hour), 1, false) + if used, known := s.QuotaSince("h1", now.Add(-90*time.Minute)); used != 5 || known { + t.Fatalf("indexed quota=(%v,%v), want (5,false)", used, known) + } + if used, known := s.QuotaSince("h1", now.Add(-30*time.Minute)); used != 4 || !known { + t.Fatalf("recent indexed quota=(%v,%v), want (4,true)", used, known) + } + reopened, err := Open(dir) + if err != nil { + t.Fatal(err) + } + if got := reopened.SchedulingSnapshot().ActiveLeases["h1"]; got != 1 { + t.Fatalf("replayed active lease index=%d, want 1", got) + } + if used, known := reopened.QuotaSince("h1", now.Add(-90*time.Minute)); used != 5 || known { + t.Fatalf("replayed indexed quota=(%v,%v), want (5,false)", used, known) + } +} + func TestReplayLegacyLeaseDerivesNonRenewableFence(t *testing.T) { dir := t.TempDir() until := time.Now().Add(time.Hour).UnixNano()