package inventory import ( "context" "errors" "fmt" "path" "sort" "strings" "time" "github.com/jackc/pgx/v5" "github.com/novox/mesh-controller/internal/chain" ) // What the controller asked to be built, and the assignments waiting for it (novox/hq issue 325, ADR 0261). // // A build's outcome was kept, and its asking was not: between a merge asking for a new module and the // outcome registering it, the mesh had no record that anything was coming, and `assign` answered "no module // of that name" for a module minutes from existing. Kept here, a build request lets `assign` say which case // it is in, and an assignment made while the build runs is kept until the build registers the module. // KeptFor is how long a build request, and an ended pending assignment, is kept. const KeptFor = 30 * 24 * time.Hour // BuildRequest is one build the controller asked for. type BuildRequest struct { // ID is the build's id, as its outcome will carry it; for a merge whose ask could not be made, an id // of the controller's making that no build carries. ID string // Repository and Seat are the source as the mesh spells it (novox/hq ADR 0111): a path on the seat's // holder, or a URL with no seat. Path is the module's directory in it, Ref the branch or commit asked. Repository, Seat, Path, Ref string // Commit is the commit that made the ask, where one did: a merge's. Commit string // For says who asked: "merge", "plan", "build" (a person), "assign". For string // NotAsked is why the ask could not be handed over; empty when it was. NotAsked string // OutcomeUnknown is why an asker that handed the build over stopped waiting for it: asked, outcome // unknown. Read as in flight until its outcome or its bound. OutcomeUnknown string At time.Time // AtTerminal says the request was asked at the controller's terminal, never through a verb (novox/hq ADR // 0266): what lets its outcome register a module from a repository the catalogue does not build it from. AtTerminal bool } // Name is the module this request is expected to register, read from its directory: the last element of // its path, or of its repository for a module at the repository's root. A guess until the build says, and // the one a person reads too: `modules/sensors` is sensors. func (a BuildRequest) Name() string { if p := strings.Trim(a.Path, "/"); p != "" && p != "." { return path.Base(p) } return strings.TrimSuffix(path.Base(strings.TrimRight(a.Repository, "/")), ".git") } // RequestOutcome is a build request beside what came of it, where anything did. type RequestOutcome struct { BuildRequest // Heard is whether an outcome of this id was taken in; Failed is its builder's words when it failed, // Module what the build said it built, HeardAt when it was taken in. Heard bool Failed string Module string HeardAt time.Time } // RecordBuildRequest keeps one build request, once it is asked. Idempotent on the id. Requests older than // KeptFor are deleted in the same act. func (i *Inventory) RecordBuildRequest(ctx context.Context, a BuildRequest) error { if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { return err } else if on { return i.chainRecordBuildRequest(ctx, c, a) } at := a.At if at.IsZero() { at = time.Now().UTC() } var notAsked *string if a.NotAsked != "" { notAsked = &a.NotAsked } if _, err := i.store.Pool().Exec(ctx, `insert into build_request (id, repository, seat, source_path, ref, commit_hash, asked_for, not_asked, asked_at, at_terminal) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) on conflict (id) do nothing`, a.ID, a.Repository, a.Seat, strings.Trim(a.Path, "/"), a.Ref, a.Commit, a.For, notAsked, at, a.AtTerminal); err != nil { return err } _, err := i.store.Pool().Exec(ctx, `delete from build_request where asked_at < $1`, time.Now().Add(-KeptFor)) return err } // BuildRequestByID is the build request kept under this id, and whether one was kept. func (i *Inventory) BuildRequestByID(ctx context.Context, id string) (BuildRequest, bool, error) { if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { return BuildRequest{}, false, err } else if on { return i.chainBuildRequestByID(ctx, c, id) } var a BuildRequest err := i.store.Pool().QueryRow(ctx, `select id, repository, seat, source_path, ref, commit_hash, asked_for, at_terminal, asked_at from build_request where id = $1`, id).Scan(&a.ID, &a.Repository, &a.Seat, &a.Path, &a.Ref, &a.Commit, &a.For, &a.AtTerminal, &a.At) if errors.Is(err, pgx.ErrNoRows) { return BuildRequest{}, false, nil } return a, err == nil, err } // MarkNotAsked says a kept build request was never handed over: the words are kept, unless its outcome was // heard first. func (i *Inventory) MarkNotAsked(ctx context.Context, id, why string) error { if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { return err } else if on { return i.chainMarkRequest(ctx, c, id, func(r *chain.Request) { r.NotAsked = why }) } if heard, err := i.outcomeHeard(ctx, id); err != nil || heard { return err } _, err := i.store.Pool().Exec(ctx, `update build_request set not_asked = $2 where id = $1`, id, why) return err } // MarkOutcomeUnknown says the asker of a build it handed over stopped waiting for it: asked, outcome unknown. // The build may still run; its outcome, when heard, is the last word. func (i *Inventory) MarkOutcomeUnknown(ctx context.Context, id, why string) error { if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { return err } else if on { return i.chainMarkRequest(ctx, c, id, func(r *chain.Request) { r.OutcomeUnknown = why }) } if heard, err := i.outcomeHeard(ctx, id); err != nil || heard { return err } _, err := i.store.Pool().Exec(ctx, `update build_request set outcome_unknown = $2 where id = $1`, id, why) return err } // outcomeHeard says a build of this id was recorded in the table: its outcome was taken in. func (i *Inventory) outcomeHeard(ctx context.Context, id string) (bool, error) { var heard bool err := i.store.Pool().QueryRow(ctx, `select exists (select 1 from build where id = $1)`, id).Scan(&heard) return heard, err } // RequestsNamed is every build request kept whose directory names the module, or whose build said it built // it, newest first, each with its outcome. func (i *Inventory) RequestsNamed(ctx context.Context, name string) ([]RequestOutcome, error) { all, err := i.requests(ctx) if err != nil { return nil, err } var out []RequestOutcome for _, a := range all { if a.Name() == name || a.Module == name { out = append(out, a) } } return out, nil } // RequestedNames is the name of every module a kept build request is expected to register: what `assign` // offers as a closest name beside the modules registered. func (i *Inventory) RequestedNames(ctx context.Context) ([]string, error) { all, err := i.requests(ctx) if err != nil { return nil, err } seen := map[string]bool{} var out []string for _, a := range all { if n := a.Name(); n != "" && !seen[n] { seen[n] = true out = append(out, n) } } return out, nil } func (i *Inventory) requests(ctx context.Context) ([]RequestOutcome, error) { if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { return nil, err } else if on { return i.chainRequests(ctx, c) } rows, err := i.store.Pool().Query(ctx, `select a.id, a.repository, a.seat, a.source_path, a.ref, a.commit_hash, a.asked_for, coalesce(a.not_asked, ''), coalesce(a.outcome_unknown, ''), a.asked_at, b.id is not null, coalesce(b.failed, ''), coalesce(b.module, ''), coalesce(b.at, a.asked_at) from build_request a left join build b on b.id = a.id order by a.asked_at desc, a.id desc`) if err != nil { return nil, err } defer rows.Close() var out []RequestOutcome for rows.Next() { var a RequestOutcome if err := rows.Scan(&a.ID, &a.Repository, &a.Seat, &a.Path, &a.Ref, &a.Commit, &a.For, &a.NotAsked, &a.OutcomeUnknown, &a.At, &a.Heard, &a.Failed, &a.Module, &a.HeardAt); err != nil { return nil, err } out = append(out, a) } return out, rows.Err() } // The states of a pending assignment. const ( PendingWaiting = chain.PendingWaiting PendingApplying = chain.PendingApplying PendingApplied = chain.PendingApplied PendingRefused = chain.PendingRefused PendingExpired = chain.PendingExpired PendingWithdrawn = chain.PendingWithdrawn ) // PendingAssignment is an assignment kept until its module is registered (ADR 0261): the value of its key in the // assignments bucket while it waits (state pending), and the record carried into the value that stands once it is made. type PendingAssignment struct { // ID is p-, made when it was recorded. ID string Node string Module string // Build is the build it waits for; Repository and Path where that build is from. Build, Repository, Path string Since time.Time State string // Note is what ended it, in the words said when it ended. Note string Settled *time.Time // Raised is when an expiry or refusal was raised as a condition; Acknowledged when a person took it // back; Cleared when its condition was cleared. Raised, Acknowledged, Cleared *time.Time } // Open is whether the pending assignment is still to be made. func (p PendingAssignment) Open() bool { return p.State == PendingWaiting || p.State == PendingApplying } // ErrAlreadyPending is a second pending assignment of one module to one machine. var ErrAlreadyPending = errors.New("that assignment is already waiting for its build") // ErrOnTheMachine is a pending assignment of a module that is on the machine already. var ErrOnTheMachine = errors.New("that module is on the machine already") func pendingOf(v chain.Assignment) PendingAssignment { return pendingRecord(v, *v.Pending) } // pendingRecord is one wait of a key, the one it holds or one it replaced. func pendingRecord(v chain.Assignment, p chain.Pending) PendingAssignment { return PendingAssignment{ID: p.ID, Node: v.Node, Module: v.Module, Build: p.Build, Repository: p.Repository, Path: p.Path, Since: p.Since, State: p.State, Note: p.Note, Settled: p.Settled, Raised: p.Raised, Acknowledged: p.Acknowledged, Cleared: p.Cleared} } // pendingValues is every value of the bucket that holds a pending assignment, on a machine the store knows (a removed // machine's is none, as the node's foreign key made it), with its key and revision. func (i *Inventory) pendingValues(ctx context.Context) ([]chain.Kept[chain.Assignment], error) { c, err := i.theChain(ctx) if err != nil { return nil, err } all, err := c.Assignments.All(ctx) if err != nil { return nil, fmt.Errorf("the pending assignments cannot be read from the bus: %w", err) } nodes, err := i.nodeNames(ctx) if err != nil { return nil, err } var out []chain.Kept[chain.Assignment] for key, k := range all { if k.Value.Pending != nil && nodes[k.Value.Node] { k.Key = key out = append(out, k) } } sort.Slice(out, func(a, b int) bool { x, y := out[a].Value.Pending, out[b].Value.Pending if !x.Since.Equal(y.Since) { return x.Since.Before(y.Since) } return x.ID < y.ID }) return out, nil } func pendingList(kept []chain.Kept[chain.Assignment], keep func(chain.Pending) bool) []PendingAssignment { var out []PendingAssignment for _, k := range kept { for _, p := range pendingWaits(k.Value) { if keep(p) { out = append(out, pendingRecord(k.Value, p)) } } } sort.SliceStable(out, func(a, b int) bool { if !out[a].Since.Equal(out[b].Since) { return out[a].Since.Before(out[b].Since) } return out[a].ID < out[b].ID }) return out } // pendingWaits is the wait a value holds and the earlier ones it carries, each without the others. func pendingWaits(v chain.Assignment) []chain.Pending { if v.Pending == nil { return nil } out := make([]chain.Pending, 0, 1+len(v.Pending.Earlier)) for _, e := range v.Pending.Earlier { e.Earlier = nil out = append(out, e) } held := *v.Pending held.Earlier = nil return append(out, held) } // changePending changes the pending assignment of an id — the wait a key holds or one it carries from before — by // compare-and-set on the key that holds it, read again when it moved: false when no key holds that id now, or change says no. func (i *Inventory) changePending(ctx context.Context, id string, change func(p *chain.Pending) bool) (bool, error) { c, err := i.theChain(ctx) if err != nil { return false, err } for try := 0; try < casTries; try++ { held, err := i.pendingValues(ctx) if err != nil { return false, err } var at *chain.Kept[chain.Assignment] var target *chain.Pending for k := range held { v := held[k].Value p := *v.Pending p.Earlier = append([]chain.Pending(nil), p.Earlier...) v.Pending = &p switch { case p.ID == id: at, target = &held[k], v.Pending default: for e := range v.Pending.Earlier { if v.Pending.Earlier[e].ID == id { at, target = &held[k], &v.Pending.Earlier[e] } } } if at != nil { held[k].Value = v break } } if at == nil { return false, nil } if !change(target) { return false, nil } if _, err := c.Assignments.Update(ctx, at.Key, at.Value, at.Revision); errors.Is(err, chain.ErrMoved) { continue } else if err != nil { return false, err } return true, nil } return false, fmt.Errorf("the pending assignment %s kept moving under the write", id) } // RecordPending keeps an assignment until its module is registered, as the value of its key (state pending). One is open per // machine and module; a key that holds the module on the machine refuses it, and one that holds an ended assignment or a // settled wait is written over. func (i *Inventory) RecordPending(ctx context.Context, p PendingAssignment) (PendingAssignment, error) { node, err := i.NodeByName(ctx, p.Node) if err != nil { return PendingAssignment{}, err } c, err := i.theChain(ctx) if err != nil { return PendingAssignment{}, err } key := assignmentKey(node.Name, p.Module) for try := 0; try < casTries; try++ { cur, found, err := c.Assignments.Get(ctx, key) if err != nil { return PendingAssignment{}, err } switch { case found && cur.Value.Stands(): return PendingAssignment{}, fmt.Errorf("%w: %s on %s", ErrOnTheMachine, p.Module, node.Name) case found && cur.Value.Pending != nil && cur.Value.State == chain.StatePending && cur.Value.Pending.Open(): return PendingAssignment{}, ErrAlreadyPending } since := time.Now().UTC() p.ID, p.Since, p.State = fmt.Sprintf("p-%d", since.UnixNano()), since, PendingWaiting p.Path = strings.Trim(p.Path, "/") value := chain.Assignment{Node: node.Name, Module: p.Module, State: chain.StatePending, Pending: &chain.Pending{ ID: p.ID, State: PendingWaiting, Build: p.Build, Repository: p.Repository, Path: p.Path, Since: since, Earlier: unansweredWaits(cur.Value, found)}} if found { _, err = c.Assignments.Update(ctx, key, value, cur.Revision) } else { _, err = c.Assignments.Create(ctx, key, value) } if errors.Is(err, chain.ErrMoved) || errors.Is(err, chain.ErrExists) { continue } if err != nil { return PendingAssignment{}, err } return p, nil } return PendingAssignment{}, fmt.Errorf("the assignment of %s to %s kept moving under the write", p.Module, node.Name) } // RepointPending makes a waiting pending assignment wait for another build. False when it no longer waits. func (i *Inventory) RepointPending(ctx context.Context, id string, build, repository, sourcePath string) (bool, error) { return i.changePending(ctx, id, func(p *chain.Pending) bool { if p.State != PendingWaiting { return false } p.Build, p.Repository, p.Path = build, repository, strings.Trim(sourcePath, "/") return true }) } // Pending is every pending assignment still waiting, oldest first, when endedSince is zero. Otherwise it is // every open one and every one ended since then. func (i *Inventory) Pending(ctx context.Context, endedSince time.Time) ([]PendingAssignment, error) { held, err := i.pendingValues(ctx) if err != nil { return nil, err } if endedSince.IsZero() { return pendingList(held, func(p chain.Pending) bool { return p.State == PendingWaiting }), nil } return pendingList(held, func(p chain.Pending) bool { return p.Open() || (p.Settled != nil && !p.Settled.Before(endedSince)) }), nil } // PendingFor is every pending assignment of one module to one machine the key still has to say, newest first: the wait it // holds now and the unanswered ones it replaced. func (i *Inventory) PendingFor(ctx context.Context, node, module string) ([]PendingAssignment, error) { held, err := i.pendingValues(ctx) if err != nil { return nil, err } var out []PendingAssignment for _, k := range held { if k.Value.Node == node && k.Value.Module == module { for _, w := range pendingWaits(k.Value) { out = append(out, pendingRecord(k.Value, w)) } } } // Newest first. sort.SliceStable(out, func(a, b int) bool { return out[a].Since.After(out[b].Since) }) return out, nil } // Stuck is every pending assignment left applying since before the time given: a controller that claimed // it and stopped before it said what came of it. func (i *Inventory) Stuck(ctx context.Context, before time.Time) ([]PendingAssignment, error) { held, err := i.pendingValues(ctx) if err != nil { return nil, err } return pendingList(held, func(p chain.Pending) bool { return p.State == PendingApplying && p.Settled != nil && p.Settled.Before(before) }), nil } // ToRaise is every pending assignment that expired or was refused and has not been raised as a condition; // ToClear every one raised and not cleared. func (i *Inventory) ToRaise(ctx context.Context) ([]PendingAssignment, error) { held, err := i.pendingValues(ctx) if err != nil { return nil, err } return pendingList(held, func(p chain.Pending) bool { return (p.State == PendingExpired || p.State == PendingRefused) && p.Raised == nil }), nil } func (i *Inventory) ToClear(ctx context.Context) ([]PendingAssignment, error) { held, err := i.pendingValues(ctx) if err != nil { return nil, err } return pendingList(held, func(p chain.Pending) bool { return p.Raised != nil && p.Cleared == nil }), nil } // MarkPending stamps a pending assignment raised, acknowledged or cleared, now. func (i *Inventory) MarkPending(ctx context.Context, id string, what string) error { if what != "raised" && what != "acknowledged" && what != "cleared" { return errors.New("a pending assignment is marked raised, acknowledged or cleared") } now := time.Now().UTC() _, err := i.changePending(ctx, id, func(p *chain.Pending) bool { mark := map[string]**time.Time{"raised": &p.Raised, "acknowledged": &p.Acknowledged, "cleared": &p.Cleared}[what] if *mark != nil { return false // the first stamp stands } *mark = &now return true }) return err } // ClaimPending takes a waiting pending assignment for making, so nothing else settles or withdraws it while // it is made. False when it no longer waits: another process took it, or it was withdrawn. func (i *Inventory) ClaimPending(ctx context.Context, id string) (bool, error) { now := time.Now().UTC() return i.changePending(ctx, id, func(p *chain.Pending) bool { if !chain.CanMove(p.State, PendingApplying) { return false } p.State, p.Settled = PendingApplying, &now return true }) } // ReleaseClaim puts a claimed pending assignment back to waiting. func (i *Inventory) ReleaseClaim(ctx context.Context, id string) error { _, err := i.changePending(ctx, id, func(p *chain.Pending) bool { if p.State != PendingApplying { return false } p.State, p.Settled = PendingWaiting, nil return true }) return err } // SettlePending ends a pending assignment in the state it is read in (from): waiting for an expiry or a // withdrawal, applying for the outcome of making it. False when it was no longer in that state. func (i *Inventory) SettlePending(ctx context.Context, id string, from, state, note string) (bool, error) { if state == PendingWaiting || state == PendingApplying { return false, errors.New("a pending assignment is settled to applied, refused, expired or withdrawn") } now := time.Now().UTC() return i.changePending(ctx, id, func(p *chain.Pending) bool { if p.State != from || !chain.CanSettle(from, state) { return false } p.State, p.Note, p.Settled = state, note, &now return true }) } // PendingKnown says which of these pending assignments are still on record: a key holds them now, or carries them as // unanswered earlier waits. func (i *Inventory) PendingKnown(ctx context.Context, ids []string) (map[string]bool, error) { held, err := i.pendingValues(ctx) if err != nil { return nil, err } want := map[string]bool{} for _, id := range ids { want[id] = true } out := map[string]bool{} for _, k := range held { for _, w := range pendingWaits(k.Value) { if want[w.ID] { out[w.ID] = true } } } return out, nil } // unansweredWaits are the waits a key holds that ended unmade and whose condition nobody has answered yet — taken back or // cleared — which a new wait replacing them carries, so the operator is not left without it (issue 1108). func unansweredWaits(v chain.Assignment, found bool) []chain.Pending { if !found || v.State != chain.StatePending { return nil } var out []chain.Pending for _, w := range pendingWaits(v) { if (w.State == PendingExpired || w.State == PendingRefused) && w.Cleared == nil && w.Acknowledged == nil { out = append(out, w) } } return out }