package inventory import ( "context" "encoding/json" "errors" "fmt" "sort" "strings" "time" "github.com/jackc/pgx/v5" "github.com/novox/mesh-controller/internal/chain" ) // Assignments on the bus (novox/hq ADR 0408, issue 1093): once the kind `assignments` is moved (chain.go) an assignment is // a key `.` of the assignments bucket, written by compare-and-set by the controller that holds the lease, // and every question the joins of the store answered is answered from the bucket. **No key is removed** (ADR 0408 ยง4): an // unassignment writes the value again with `ended` set, and a value with `ended` or a `state` is no assignment to any reader // here (chain.Assignment.Stands). No `pending` state is written yet (issue 1094). // KindAssignments is the kind of the chain that is each module assigned to a node, with the build it runs there. func assignmentKey(node, module string) string { return node + "." + module } // assignRow is one assignment as the joins of the store read it. type assignRow struct { Node, Module string Assigned time.Time } // assignments is every assignment, by node then module, read from the assignments bucket. // A key whose value names no time assigned is a module only pulled in by another (issue 765), and one that has ended or // waits in a state is gone: no assignment. One on a machine the store does not know is none either, as the store took a // removed machine's. func (i *Inventory) assignments(ctx context.Context) ([]assignRow, 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 assignments cannot be read from the bus: %w", err) } nodes, err := i.nodeNames(ctx) if err != nil { return nil, err } var out []assignRow for _, k := range all { if k.Value.Assigned.IsZero() || !k.Value.Stands() || !nodes[k.Value.Node] { continue } out = append(out, assignRow{k.Value.Node, k.Value.Module, k.Value.Assigned}) } sort.Slice(out, func(a, b int) bool { if out[a].Node != out[b].Node { return out[a].Node < out[b].Node } return out[a].Module < out[b].Module }) return out, nil } func (i *Inventory) nodeNames(ctx context.Context) (map[string]bool, error) { rows, err := i.store.Pool().Query(ctx, `select name from node`) if err != nil { return nil, err } defer rows.Close() out := map[string]bool{} for rows.Next() { var name string if err := rows.Scan(&name); err != nil { return nil, err } out[name] = true } return out, rows.Err() } // assignedNodes is the machines a module is assigned to, sorted. func (i *Inventory) assignedNodes(ctx context.Context, module string) ([]string, error) { rows, err := i.assignments(ctx) if err != nil { return nil, err } var out []string for _, a := range rows { if a.Module == module { out = append(out, a.Node) } } return out, nil } // assignedModules is the modules assigned to a machine, sorted. func (i *Inventory) assignedModules(ctx context.Context, node string) ([]string, error) { rows, err := i.assignments(ctx) if err != nil { return nil, err } var out []string for _, a := range rows { if a.Node == node { out = append(out, a.Module) } } return out, nil } // isAssigned says a module is assigned to a machine. func (i *Inventory) isAssigned(ctx context.Context, node, module string) (bool, error) { rows, err := i.assignments(ctx) if err != nil { return false, err } for _, a := range rows { if a.Node == node && a.Module == module { return true, nil } } return false, nil } // raiseGeneration raises the assignment generation, as the store's triggers do for a change of its table (migration // 0094): in the chain nothing in the store changes with an assignment, so the writer says it. func (i *Inventory) raiseGeneration(ctx context.Context) error { _, err := i.store.Pool().Exec(ctx, `update assignment_generation set generation = generation + 1`) return err } // chainAssign puts a module on a machine, and says whether that is new. func (i *Inventory) chainAssign(ctx context.Context, c *chain.Chain, node Node, module string) (bool, error) { var known bool if err := i.store.Pool().QueryRow(ctx, `select exists (select 1 from module where name = $1)`, module).Scan(&known); err != nil { return false, err } if !known { return false, fmt.Errorf("%w: %s", ErrNoSuchModule, module) } key := assignmentKey(node.Name, module) for try := 0; try < casTries; try++ { cur, found, err := c.Assignments.Get(ctx, key) if err != nil { return false, err } if found && !cur.Value.Assigned.IsZero() && cur.Value.Stands() { return false, nil // one assignment of a module per machine: a repeat is absorbed (novox/hq ADR 0115) } now := time.Now().UTC() fresh := chain.Assignment{Node: node.Name, Module: module, Assigned: now} switch { case found && cur.Value.Stands(): // A module only pulled in by another becomes an assignment, keeping what it holds. cur.Value.Assigned, cur.Value.PulledIn = now, false _, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision) case found: // One that ended is assigned again, and one that was pending is made: a fresh value over it, the earlier one // staying in history. The wait is carried into the value that stands (issue 1094). fresh.Pending = carriedPending(cur.Value, now) _, err = c.Assignments.Update(ctx, key, fresh, cur.Revision) default: _, err = c.Assignments.Create(ctx, key, fresh) } if errors.Is(err, chain.ErrMoved) || errors.Is(err, chain.ErrExists) { continue } if err != nil { return false, fmt.Errorf("%s is not assigned to %s on the bus: %w", module, node.Name, err) } if err := i.raiseGeneration(ctx); err != nil { return true, err } return true, nil } return false, fmt.Errorf("the assignment of %s to %s kept moving under the write", module, node.Name) } // chainUnassign takes a module off a machine: the value is written again with `ended`, the key stays. func (i *Inventory) chainUnassign(ctx context.Context, c *chain.Chain, node Node, module string) error { key := assignmentKey(node.Name, module) for try := 0; try < casTries; try++ { cur, found, err := c.Assignments.Get(ctx, key) if err != nil { return err } if !found || cur.Value.Assigned.IsZero() || !cur.Value.Stands() { return fmt.Errorf("%s is not assigned to %s", module, node.Name) } cur.Value.Ended = time.Now().UTC() if _, err := c.Assignments.Update(ctx, key, cur.Value, cur.Revision); errors.Is(err, chain.ErrMoved) { continue } else if err != nil { return fmt.Errorf("%s is not taken off %s on the bus: %w", module, node.Name, err) } return i.raiseGeneration(ctx) } return fmt.Errorf("the assignment of %s to %s kept moving under the write", module, node.Name) } // chainAwaitingSince is AwaitingSince over the bucket: the oldest assignment on the machine made after it was last sent. func (i *Inventory) chainAwaitingSince(ctx context.Context, name string) (time.Time, error) { var sentAt *time.Time var created time.Time err := i.store.Pool().QueryRow(ctx, `select sent_at, created from node where name = $1`, name).Scan(&sentAt, &created) if errors.Is(err, pgx.ErrNoRows) { return time.Time{}, fmt.Errorf("%w: %s", ErrNoSuchNode, name) } if err != nil { return time.Time{}, err } rows, err := i.assignments(ctx) if err != nil { return time.Time{}, err } var oldest time.Time for _, a := range rows { if a.Node == name && (sentAt == nil || a.Assigned.After(*sentAt)) && (oldest.IsZero() || a.Assigned.Before(oldest)) { oldest = a.Assigned } } switch { case !oldest.IsZero(): return oldest, nil case sentAt != nil: return *sentAt, nil } return created, nil } // KeepAssignmentPlans writes the provisioner's plan of each assignment into its value on the bus, and a value of its own for // each module only pulled in by another on a machine (issue 765), and ends the value of one that is no longer pulled in. The // plans are derived from what each machine was last sent, not part of the record of what is assigned, so they are read // again on a tick and written where they differ: **by compare-and-set on the one field**, never laying over the build, commit // and rollout another write put in the same value, and never ending an assignment. A plan the provisioner's rules refuse is // refused by name and nothing is written (AssignmentValues). func (i *Inventory) KeepAssignmentPlans(ctx context.Context) (written, removed int, err error) { c, err := i.theChain(ctx) if err != nil { return 0, 0, err } var plans map[string]Provisioning if i.provisioningPlans != nil { if plans, err = i.provisioningPlans(ctx); err != nil { return 0, 0, fmt.Errorf("composing the provisioner's plans: %w", err) } } assigned, err := i.assignments(ctx) if err != nil { return 0, 0, err } rows := make([]AssignmentMirrored, 0, len(assigned)) for _, a := range assigned { rows = append(rows, AssignmentMirrored{Node: a.Node, Module: a.Module, Assigned: a.Assigned.UTC()}) } want, err := AssignmentValues(rows, plans) if err != nil { return 0, 0, err } have, err := c.Assignments.All(ctx) if err != nil { return 0, 0, fmt.Errorf("the assignments cannot be read from the bus: %w", err) } for key, body := range want { var next chain.Assignment if err := json.Unmarshal(body, &next); err != nil { return written, removed, err } cur, has := have[key] switch keepStep(has, cur.Value, next) { case keepSkip: continue case keepCreate: next.Assigned, next.PulledIn = time.Time{}, true _, err = c.Assignments.Create(ctx, key, next) case keepFresh: // A value that ended, pulled in again: a fresh one over it, the ended one staying in history. next.Assigned, next.PulledIn = time.Time{}, true _, err = c.Assignments.Update(ctx, key, next, cur.Revision) default: cur.Value.RequiredBy, cur.Value.Provisioning = next.RequiredBy, next.Provisioning _, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision) } if errors.Is(err, chain.ErrMoved) || errors.Is(err, chain.ErrExists) { continue // written meanwhile; the next pass reads it } if err != nil { return written, removed, fmt.Errorf("the plan of %s is not written: %w", key, err) } written++ } for key, cur := range have { if _, wanted := want[key]; wanted || !cur.Value.Assigned.IsZero() || !cur.Value.Stands() || cur.Value.Commit != "" { continue // one a send still carries is on the machine, and ends when a send does not carry it any more } // A value of a module no longer pulled in, which no assignment names: it ends, and stays. cur.Value.Ended = time.Now().UTC() if _, err := c.Assignments.Update(ctx, key, cur.Value, cur.Revision); errors.Is(err, chain.ErrMoved) { continue } else if err != nil { return written, removed, fmt.Errorf("the plan of %s is not ended: %w", key, err) } removed++ } return written, removed, nil } // sameJSON says two encoded values are the same bytes, an absent value being a null one. func sameJSON(a, b json.RawMessage) bool { trim := func(x json.RawMessage) string { if s := strings.TrimSpace(string(x)); s != "null" { return s } return "" } return trim(a) == trim(b) } // KeptValue is an assignment's value as the bucket holds it, encoded, at its revision. type KeptValue struct { Value []byte Revision uint64 } // KeptAssignments is every key of the assignments bucket that holds a value that stands. What the serving controller // compares from one look to the next to say a key changed (`assignment-changed`). func (i *Inventory) KeptAssignments(ctx context.Context) (map[string]KeptValue, error) { c, err := i.theChain(ctx) if err != nil { return nil, err } all, err := c.Assignments.All(ctx) if err != nil { return nil, err } out := make(map[string]KeptValue, len(all)) for key, k := range all { if !k.Value.Stands() { continue // an ended value is no assignment: for what is said of it, the key has gone } body, err := json.Marshal(k.Value) if err != nil { return nil, err } out[key] = KeptValue{Value: body, Revision: k.Revision} } return out, nil } // What a pass that keeps the plans does with one key. const ( keepSkip = "skip" keepCreate = "create" keepFresh = "fresh" keepField = "field" ) // keepStep decides what the pass does with a key from what it wants there (read from the assignments) and what the bucket // holds (read a moment later). **It never makes or revives an assignment**: that is Assign's, and an unassignment between the // two reads leaves a key that ended while the pass still wants it as an assignment, which would come back as a value // pulled in by another. Only a key with no time assigned, on both sides, is made or made again here. func keepStep(has bool, cur, next chain.Assignment) string { switch { case !has && !next.Assigned.IsZero(): return keepSkip // assigned since the bucket was read: its own write is not here yet case !has: return keepCreate case cur.State == chain.StatePending: return keepSkip // a wait for a build is not pulled in by a plan case !cur.Stands(): if !cur.Assigned.IsZero() || !next.Assigned.IsZero() { return keepSkip // an assignment that ended is not pulled in again by a plan } return keepFresh case sameJSON(cur.Provisioning, next.Provisioning) && cur.RequiredBy == next.RequiredBy: return keepSkip } return keepField } // carriedPending is the wait a value that does not stand carries into the assignment made over it: a wait that was applying // stays applying, for the one making it settles it applied; one still waiting is made by hand, and says so. Nothing for a // value that held no wait, or whose key is not one that was pending. func carriedPending(was chain.Assignment, now time.Time) *chain.Pending { if was.Pending == nil || was.State != chain.StatePending { return nil } p := *was.Pending if p.State == chain.PendingWaiting { p.State, p.Note, p.Settled = chain.PendingApplied, "assigned by hand", &now } return &p } // AssignmentValue is one key of the assignments bucket as it is held, standing or not: one read of an assignment. func (i *Inventory) AssignmentValue(ctx context.Context, node, module string) (chain.Assignment, bool, error) { c, err := i.theChain(ctx) if err != nil { return chain.Assignment{}, false, err } k, found, err := c.Assignments.Get(ctx, assignmentKey(node, module)) return k.Value, found, err }