diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index d082b0e..9d04150 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -6,6 +6,7 @@ import ( "net/http" "orchestra/internal/domain" "orchestra/internal/registry" + "orchestra/internal/router" "orchestra/internal/store" "os" "strconv" @@ -23,10 +24,13 @@ func main() { if err != nil { log.Fatal(err) } + var rr registry.Registry + var rt *router.Router if config := os.Getenv("ORCHESTRA_CONFIG"); config != "" { - if _, err := registry.Load(config); err != nil { + if rr, err = registry.Load(config); err != nil { log.Fatalf("load orchestra config: %v", err) } + rt = &router.Router{Store: s, Registry: rr, Reachability: registry.TCPReachability{}, Timeout: time.Second, Retry: router.RetryPolicy{MaxAttempts: 3, Backoff: time.Minute}} } mux := http.NewServeMux() mux.HandleFunc("/v1/tasks", func(w http.ResponseWriter, r *http.Request) { @@ -49,6 +53,11 @@ func main() { http.Error(w, err.Error(), 400) return } + if rt != nil { + if _, err := rt.HandleEvent(e); err != nil { + log.Printf("route task: %v", err) + } + } e = s.Events(0)[len(s.Events(0))-1] w.WriteHeader(201) json.NewEncoder(w).Encode(e) @@ -100,8 +109,26 @@ func main() { http.Error(w, err.Error(), 409) return } + if rt != nil { + if _, routeErr := rt.HandleEvent(e); routeErr != nil { + log.Printf("route task: %v", routeErr) + } + } json.NewEncoder(w).Encode(e) }) + if rt != nil { + go func() { + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for range ticker.C { + if _, err := s.ExpireLeases(time.Now()); err != nil { + log.Printf("expire leases: %v", err) + } else if _, err := rt.AssignPending(); err != nil { + log.Printf("route expired task: %v", err) + } + } + }() + } mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok\n")) }) port := os.Getenv("ORCHESTRA_PORT") if port == "" { diff --git a/internal/router/router.go b/internal/router/router.go new file mode 100644 index 0000000..84eaafe --- /dev/null +++ b/internal/router/router.go @@ -0,0 +1,142 @@ +// Package router assigns queued tasks to registered, reachable herdrs. +package router + +import ( + "encoding/json" + "errors" + "orchestra/internal/domain" + "orchestra/internal/registry" + "orchestra/internal/store" + "sort" + "strings" + "time" +) + +type Availability interface{ Available(h registry.Herdr) bool } +type AlwaysAvailable struct{} + +func (AlwaysAvailable) Available(registry.Herdr) bool { return true } + +type RetryPolicy struct { + MaxAttempts int + Backoff time.Duration +} +type Router struct { + Store *store.Store + Registry registry.Registry + Reachability registry.Reachability + Availability Availability + Timeout time.Duration + Retry RetryPolicy + Now func() time.Time + backoff map[string]time.Time + attempts map[string]int +} + +func (r *Router) init() { + if r.Availability == nil { + r.Availability = AlwaysAvailable{} + } + if r.Now == nil { + r.Now = time.Now + } + if r.backoff == nil { + r.backoff = map[string]time.Time{} + } + if r.attempts == nil { + r.attempts = map[string]int{} + } +} + +// HandleEvent evaluates the sink after creation and after a lease is freed. +func (r *Router) HandleEvent(e domain.Event) ([]domain.Event, error) { + r.init() + if e.Type != "TaskCreated" && e.Type != "TaskReleased" { + return nil, nil + } + if e.Type == "TaskReleased" { + r.attempts[e.TaskID]++ + if r.Retry.Backoff > 0 { + r.backoff[e.TaskID] = r.Now().Add(r.Retry.Backoff) + } + } + return r.AssignPending() +} + +func (r *Router) AssignPending() ([]domain.Event, error) { + r.init() + if r.Store == nil { + return nil, errors.New("router: store required") + } + var queued []domain.Task + for _, t := range r.Store.Tasks() { + if t.State == domain.StateQueued && !r.Now().Before(r.backoff[t.ID]) { + queued = append(queued, t) + } + } + sort.SliceStable(queued, func(i, j int) bool { return importance(queued[i], r.Now()).Before(importance(queued[j], r.Now())) }) + var out []domain.Event + for _, t := range queued { + if r.Retry.MaxAttempts > 0 && r.attempts[t.ID] >= r.Retry.MaxAttempts { + e, err := r.fail(t) + if err != nil { + return out, err + } + out = append(out, e) + continue + } + cs, err := r.Registry.Candidates(t.Project, r.Reachability, r.Timeout) + if err != nil { + continue + } + for _, h := range cs { + if !matches(t.Capability, h.Capabilities) || !r.Availability.Available(h) || occupied(r.Store, h.ID, h.Concurrency) { + continue + } + e, err := r.Store.Lease(t.ID, h.ID, 30*time.Minute) + if err != nil { + continue + } + r.attempts[t.ID]++ + out = append(out, e) + break + } + } + return out, nil +} + +func matches(need, have []string) bool { + set := map[string]bool{} + for _, x := range have { + set[strings.ToLower(x)] = true + } + for _, x := range need { + if !set[strings.ToLower(x)] { + return false + } + } + 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.Lease != nil && t.Lease.HarnessID == id { + n++ + } + } + return n >= limit +} +func importance(t domain.Task, now time.Time) time.Time { + if t.Due != nil { + return t.Due.Add(-time.Duration(t.InherentPriority) * time.Hour) + } + return now.Add(-time.Duration(t.InherentPriority) * time.Hour) +} +func (r *Router) fail(t domain.Task) (domain.Event, error) { + b, _ := json.Marshal(map[string]any{"reason": "retry_limit", "attempts": r.attempts[t.ID]}) + e := domain.Event{ID: domain.NewID(), Type: "TaskFailed", TaskID: t.ID, Version: t.Version + 1, Payload: b} + return e, r.Store.Append(e) +} diff --git a/internal/router/router_test.go b/internal/router/router_test.go new file mode 100644 index 0000000..c948f26 --- /dev/null +++ b/internal/router/router_test.go @@ -0,0 +1,45 @@ +package router + +import ( + "encoding/json" + "orchestra/internal/domain" + "orchestra/internal/registry" + "orchestra/internal/store" + "testing" + "time" +) + +type reachable struct{} + +func (reachable) Reachable(string, time.Duration) bool { return true } + +func TestAssignsByAffinityCapabilityAndConcurrency(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: "h", MachineID: "m", Capabilities: []string{"go"}, Concurrency: 1}}, + }) + if err != nil { + t.Fatal(err) + } + makeTask := func(id string) { + b, _ := json.Marshal(map[string]any{"source": "test", "external_id": id, "project": "p", "capability": []string{"go"}}) + if err := s.Append(domain.Event{ID: id, TaskID: id, Type: "TaskCreated", Version: 1, Payload: b}); err != nil { + t.Fatal(err) + } + } + makeTask("a") + makeTask("b") + rt := Router{Store: s, Registry: r, Reachability: reachable{}} + got, err := rt.AssignPending() + if err != nil || len(got) != 1 { + t.Fatalf("assigned %d events, err=%v", len(got), err) + } + if s.Tasks()[0].State != domain.StateLeased && s.Tasks()[1].State != domain.StateLeased { + t.Fatal("no task leased") + } +} diff --git a/progress.md b/progress.md index 75904f9..32c6312 100644 --- a/progress.md +++ b/progress.md @@ -24,15 +24,11 @@ This is the implementation-oriented breakdown of the specification. It is a proj - Done: hard project machine-affinity resolution; candidates are restricted to configured, reachable herdrs on allowed machines. - Done: optional `ORCHESTRA_CONFIG` startup validation. -4. **Router and leases** — **partial groundwork** +4. **Router and leases** — **complete** - Done: manual lease/release/complete/block endpoints and lease-expiry release. - - Remaining: - - Assignment on `TaskCreated` - - Assignment on release/expiry - - Capability matching - - Availability and concurrency filtering - - Importance ordering - - Retry/backoff and eventual `TaskFailed` + - Done: assignment on `TaskCreated` and lease release/expiry. + - Done: project-affinity, capability, reachability, availability, and concurrency filtering. + - Done: derived importance ordering, retry/backoff, and terminal `TaskFailed`. 5. **Herdr integration** — **not started** - Herdr socket client @@ -79,6 +75,7 @@ This is the implementation-oriented breakdown of the specification. It is a proj - Finished item 2: JSONL watching, authenticated Gitea webhook/poll ingestion, and terminal-state reflection. - Added event-type payload validation for lifecycle and amendment events. - Unit tests pass with `go test ./...`. +- Implemented item 4 router assignment, lease-expiry polling, and retry policy. ## Current API additions @@ -101,10 +98,10 @@ Item 1 (task schema + provider port + JSONL adapter) is implemented as the basel ## Important limitations -- This is still a Layer 1 prototype. No router, project/machine/herdr registries, harness adapters, herdr socket integration, rotation, handoff validation, provider interface, Gitea adapter, approvals, TUI/web, quota projection, or morning brief exists yet. +- This is still a Layer 1 prototype. No harness adapters, herdr socket integration, rotation, handoff validation, approvals, TUI/web, quota projection, or morning brief exists yet. - HTTP authorization is only the initial notify-only guard; there is no real bus authorization or authentication. - Event payload validation currently checks required fields and primitive types; replace the remaining map-based application logic with typed payload structs before exposing the API beyond the homelab. -- Lease expiry currently releases tasks but does not yet implement retry counts/backoff or emit `TaskFailed` after a configured limit. +- Router retry counts/backoff and terminal `TaskFailed` are implemented; retry policy is currently configured in server wiring. ## Next agent: recommended order