Fold a pending assignment into the value of its assignment key (issue 1094)

A wait for a build is the value of <node>.<module> with state pending and a pending object (id p-<unix nanoseconds>, a string), so every reader that skips a value with a state skips it, and the controller's own lists skip it through Stands. The transitions are compare-and-set on the key's revision (claim, release, settle, repoint, mark); assigning over a wait carries it into the value that stands, applied when it was waiting; a plan pass never overwrites it; one open wait per machine and module, and a key that holds the module refuses a new one. The table is no longer read or written, ForgetEndedPending is gone (no key is removed), and a wait of a machine the store dropped is none. Callers change type only; their conditions are the next issue.
This commit is contained in:
2026-10-18 05:17:30 +02:00
parent 0149a3d7de
commit 88ea361bf6
11 changed files with 620 additions and 147 deletions
+18
View File
@@ -147,3 +147,21 @@ func (i *Inventory) BeginSwitchForTest(kind string) {
i.moved.switching.Store(true)
}
}
// theChain is the chain this inventory writes and reads, opened if a command has not yet: for what has no table of its own
// to be read from while it is moved (a pending assignment is a value of the assignments bucket from the start).
func (i *Inventory) theChain(ctx context.Context) (*chain.Chain, error) {
i.moved.mu.Lock()
defer i.moved.mu.Unlock()
if i.chain == nil && i.moved.open != nil {
c, err := i.moved.open(ctx)
if err != nil {
return nil, fmt.Errorf("%w: %v", ErrNoChain, err)
}
i.chain = c
}
if i.chain == nil {
return nil, ErrNoChain
}
return i.chain, nil
}
+19 -1
View File
@@ -175,7 +175,9 @@ func (i *Inventory) chainAssign(ctx context.Context, c *chain.Chain, node Node,
cur.Value.Assigned = now
_, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision)
case found:
// One that ended is assigned again: a fresh value over it, the ended one staying in history.
// 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)
@@ -384,6 +386,8 @@ func keepStep(has bool, cur, next chain.Assignment) string {
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
@@ -394,3 +398,17 @@ func keepStep(has bool, cur, next chain.Assignment) string {
}
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
}
@@ -51,18 +51,21 @@ func TestAPassThatKeepsThePlansNeverRevivesAnEndedAssignment(t *testing.T) {
ended := chain.Assignment{Node: "a", Module: "m", Assigned: at, Ended: at.Add(time.Minute)}
pulledIn := chain.Assignment{Node: "a", Module: "p", RequiredBy: "m", Provisioning: plan}
pulledInEnded := chain.Assignment{Node: "a", Module: "p", Ended: at}
waiting := chain.Assignment{Node: "a", Module: "p", State: chain.StatePending,
Pending: &chain.Pending{ID: "p-1", State: chain.PendingWaiting}}
for name, tc := range map[string]struct {
has bool
cur, next chain.Assignment
want string
}{
"the interleaving: wanted assigned, ended since": {true, ended, assigned, keepSkip},
"wanted assigned, not in the bucket yet": {false, chain.Assignment{}, assigned, keepSkip},
"a pulled-in module, not in the bucket": {false, chain.Assignment{}, pulledIn, keepCreate},
"a pulled-in module that ended, pulled again": {true, pulledInEnded, pulledIn, keepFresh},
"an assignment ended, a plan for it": {true, ended, chain.Assignment{Node: "a", Module: "m", RequiredBy: "x", Provisioning: plan}, keepSkip},
"the plan is the same": {true, assigned, assigned, keepSkip},
"the plan changed": {true, chain.Assignment{Node: "a", Module: "m", Assigned: at}, assigned, keepField},
"the interleaving: wanted assigned, ended since": {true, ended, assigned, keepSkip},
"wanted assigned, not in the bucket yet": {false, chain.Assignment{}, assigned, keepSkip},
"a pulled-in module, not in the bucket": {false, chain.Assignment{}, pulledIn, keepCreate},
"a pulled-in module that ended, pulled again": {true, pulledInEnded, pulledIn, keepFresh},
"an assignment ended, a plan for it": {true, ended, chain.Assignment{Node: "a", Module: "m", RequiredBy: "x", Provisioning: plan}, keepSkip},
"a wait for a build, a plan for the module pulled in": {true, waiting, pulledIn, keepSkip},
"the plan is the same": {true, assigned, assigned, keepSkip},
"the plan changed": {true, chain.Assignment{Node: "a", Module: "m", Assigned: at}, assigned, keepField},
} {
if got := keepStep(tc.has, tc.cur, tc.next); got != tc.want {
t.Errorf("%s: %s, want %s", name, got, tc.want)
+221
View File
@@ -0,0 +1,221 @@
//go:build integration
package inventory
import (
"errors"
"fmt"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/novox/mesh-controller/internal/chain"
)
// **A pending assignment is the value of the key of the assignment it waits for** (ADR 0261, ADR 0408, issue 1094): state
// pending, the wait in its `pending` object under an id that is a string, no table, and one open wait per machine and module.
func TestAPendingAssignmentIsAKeyOfTheAssignmentsBucket(t *testing.T) {
inv, node := aNodeWithModules(t, "web")
ctx := t.Context()
p, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "build-1", Repository: "novox/mesh-catalog",
Path: "/modules/sensors/"})
if err != nil {
t.Fatal(err)
}
if want := fmt.Sprintf("p-%d", p.Since.UnixNano()); p.ID != want || p.State != PendingWaiting || p.Path != "modules/sensors" {
t.Fatalf("recorded %+v, want id %s", p, want)
}
kept, found, err := inv.chain.Assignments.Get(ctx, "anchor.sensors")
if err != nil || !found || kept.Value.State != chain.StatePending || kept.Value.Stands() || !kept.Value.Assigned.IsZero() ||
kept.Value.Pending == nil || kept.Value.Pending.ID != p.ID || kept.Value.Pending.Build != "build-1" {
t.Fatalf("the key holds %+v (found %v, %v)", kept.Value, found, err)
}
rows, err := inv.Pending(ctx, time.Time{})
if err != nil || len(rows) != 1 || rows[0].ID != p.ID || rows[0].Node != "anchor" || rows[0].Build != "build-1" {
t.Fatalf("waiting: %+v %v", rows, err)
}
if _, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "build-2"}); !errors.Is(err, ErrAlreadyPending) {
t.Fatalf("a second open wait: %v", err)
}
if _, err := inv.Assign(ctx, node, "web"); err != nil {
t.Fatal(err)
}
if _, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "web", Build: "build-3"}); !errors.Is(err, ErrOnTheMachine) {
t.Fatalf("a wait for a module that is on the machine: %v", err)
}
// Written over once it ended: a settled wait is replaced by a new one under a new id.
if ok, err := inv.SettlePending(ctx, p.ID, PendingWaiting, PendingExpired, "boom"); err != nil || !ok {
t.Fatalf("settled %v %v", ok, err)
}
next, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "build-2"})
if err != nil || next.ID == p.ID {
t.Fatalf("a wait after one ended: %+v %v", next, err)
}
if known, _ := inv.PendingKnown(ctx, []string{p.ID, next.ID}); known[p.ID] || !known[next.ID] {
t.Fatalf("known %v: the key holds the newer one now", known)
}
if history, _ := inv.chain.Assignments.History(ctx, "anchor.sensors"); len(history) != 2 {
t.Fatalf("the earlier wait is in the key's history: %d values", len(history))
}
if n := rowsInTheTable(t, inv, "pending_assignment"); n != "0" {
t.Fatalf("the table took %s pending assignments", n)
}
}
// **A pending value is skipped by what reads assignments** (issue 1094): the machine is assigned nothing by it, the module
// runs nowhere, and a module only waiting for a build is not forgotten as assigned.
func TestARecordedPendingIsSkippedByAssignedAndTheJoins(t *testing.T) {
inv, node := aNodeWithModules(t, "web")
ctx := t.Context()
if _, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "web", Build: "b-1"}); err != nil {
t.Fatal(err)
}
if on, _ := inv.Assigned(ctx, node); len(on) != 0 {
t.Fatalf("assigned %v", on)
}
if running, _ := inv.Running(ctx, "web"); len(running) != 0 {
t.Fatalf("running %v", running)
}
if runs, _ := inv.AssignmentRuns(ctx, node); len(runs) != 0 {
t.Fatalf("runs %v", runs)
}
entries, _ := inv.Catalogued(ctx)
for _, e := range entries {
if e.Manifest.Module == "web" && len(e.On) != 0 {
t.Fatalf("web is on %v", e.On)
}
}
if err := inv.ForgetModule(ctx, "web"); errors.Is(err, ErrStillAssigned) {
t.Fatalf("a module only waiting is assigned: %v", err)
}
}
// **Two claims of one wait make one, and the other sees false** (issue 1094): by compare-and-set on the key's revision.
func TestTwoClaimsOfOnePendingMakeOneAndTheOtherSeesFalse(t *testing.T) {
inv, node := aNodeWithModules(t)
ctx := t.Context()
p, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "b-1"})
if err != nil {
t.Fatal(err)
}
var won atomic.Int32
var wg sync.WaitGroup
for k := 0; k < 8; k++ {
wg.Add(1)
go func() {
defer wg.Done()
if ok, err := inv.ClaimPending(ctx, p.ID); err == nil && ok {
won.Add(1)
}
}()
}
wg.Wait()
if won.Load() != 1 {
t.Fatalf("%d claims won", won.Load())
}
if again, err := inv.ClaimPending(ctx, p.ID); err != nil || again {
t.Fatalf("a claim of a claimed wait: %v %v", again, err)
}
if rows, _ := inv.Stuck(ctx, time.Now().Add(time.Minute)); len(rows) != 1 || rows[0].State != PendingApplying || rows[0].Settled == nil {
t.Fatalf("stuck %+v", rows)
}
if err := inv.ReleaseClaim(ctx, p.ID); err != nil {
t.Fatal(err)
}
if rows, _ := inv.Pending(ctx, time.Time{}); len(rows) != 1 || rows[0].Settled != nil {
t.Fatalf("released %+v", rows)
}
// Only a wait that waits is withdrawn or expired, only a claim is applied.
if ok, _ := inv.SettlePending(ctx, p.ID, PendingWaiting, PendingApplied, "x"); ok {
t.Fatal("a wait was applied")
}
if ok, _ := inv.SettlePending(ctx, p.ID, PendingApplying, PendingRefused, "x"); ok {
t.Fatal("a wait that is not claimed was refused as a claim")
}
if ok, err := inv.RepointPending(ctx, p.ID, "b-2", "novox/other", "x/"); err != nil || !ok {
t.Fatalf("repoint %v %v", ok, err)
}
if rows, _ := inv.Pending(ctx, time.Time{}); rows[0].Build != "b-2" || rows[0].Path != "x" {
t.Fatalf("repointed %+v", rows)
}
}
// **Assigning over a wait carries the wait into the value that stands** (issue 1094): the claim stays applying for the one
// making it to settle applied, and one still waiting is made by hand, and says so.
func TestAssignOverAPendingKeepsItsRecordAsApplied(t *testing.T) {
inv, node := aNodeWithModules(t, "web", "cache")
ctx := t.Context()
claimed, _ := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "web", Build: "b-1"})
if ok, err := inv.ClaimPending(ctx, claimed.ID); err != nil || !ok {
t.Fatal(ok, err)
}
if added, err := inv.Assign(ctx, node, "web"); err != nil || !added {
t.Fatalf("assigned %v %v", added, err)
}
standing, _, _ := inv.chain.Assignments.Get(ctx, "anchor.web")
if !standing.Value.Stands() || standing.Value.Assigned.IsZero() || standing.Value.Pending == nil ||
standing.Value.Pending.State != PendingApplying {
t.Fatalf("the value that stands: %+v", standing.Value)
}
if ok, err := inv.SettlePending(ctx, claimed.ID, PendingApplying, PendingApplied, "made"); err != nil || !ok {
t.Fatalf("settled %v %v", ok, err)
}
if on, _ := inv.Assigned(ctx, node); len(on) != 1 || on[0] != "web" {
t.Fatalf("assigned %v", on)
}
rows, _ := inv.Pending(ctx, time.Now().Add(-time.Hour))
if len(rows) != 1 || rows[0].State != PendingApplied || rows[0].Note != "made" {
t.Fatalf("what became of the wait: %+v", rows)
}
waiting, _ := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "cache", Build: "b-2"})
if _, err := inv.Assign(ctx, node, "cache"); err != nil {
t.Fatal(err)
}
got, _ := inv.PendingFor(ctx, node, "cache")
if len(got) != 1 || got[0].ID != waiting.ID || got[0].State != PendingApplied || !strings.Contains(got[0].Note, "by hand") {
t.Fatalf("a wait assigned by hand: %+v", got)
}
}
// **Marks are stamped once, and a wait of a machine the store dropped is no wait** (issue 1094).
func TestAPendingIsMarkedOnceAndGoesWithItsMachine(t *testing.T) {
inv, node := aNodeWithModules(t)
ctx := t.Context()
p, _ := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "b-1"})
if _, err := inv.SettlePending(ctx, p.ID, PendingWaiting, PendingExpired, "boom"); err != nil {
t.Fatal(err)
}
if rows, _ := inv.ToRaise(ctx); len(rows) != 1 {
t.Fatalf("to raise %+v", rows)
}
if err := inv.MarkPending(ctx, p.ID, "raised"); err != nil {
t.Fatal(err)
}
first, _ := inv.Pending(ctx, time.Now().Add(-time.Hour))
time.Sleep(2 * time.Millisecond)
if err := inv.MarkPending(ctx, p.ID, "raised"); err != nil {
t.Fatal(err)
}
second, _ := inv.Pending(ctx, time.Now().Add(-time.Hour))
if !first[0].Raised.Equal(*second[0].Raised) {
t.Fatalf("the first stamp did not stand: %v then %v", first[0].Raised, second[0].Raised)
}
if err := inv.MarkPending(ctx, p.ID, "frobnicated"); err == nil {
t.Fatal("a mark nobody knows")
}
if rows, _ := inv.ToRaise(ctx); len(rows) != 0 {
t.Fatalf("to raise %+v", rows)
}
if rows, _ := inv.ToClear(ctx); len(rows) != 1 {
t.Fatalf("to clear %+v", rows)
}
if _, err := inv.RemoveNodeForTest(ctx, node); err != nil {
t.Fatal(err)
}
if rows, _ := inv.Pending(ctx, time.Unix(0, 0)); len(rows) != 0 {
t.Fatalf("a wait outlived its machine: %+v", rows)
}
}
+219 -122
View File
@@ -3,7 +3,9 @@ package inventory
import (
"context"
"errors"
"fmt"
"path"
"sort"
"strings"
"time"
@@ -214,17 +216,19 @@ func (i *Inventory) requests(ctx context.Context) ([]RequestOutcome, error) {
// The states of a pending assignment.
const (
PendingWaiting = "waiting"
PendingApplying = "applying"
PendingApplied = "applied"
PendingRefused = "refused"
PendingExpired = "expired"
PendingWithdrawn = "withdrawn"
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).
// 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 int64
// ID is p-<unix nanoseconds>, 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.
@@ -247,195 +251,288 @@ func (p PendingAssignment) Open() bool {
// ErrAlreadyPending is a second pending assignment of one module to one machine.
var ErrAlreadyPending = errors.New("that assignment is already waiting for its build")
const pendingColumns = `p.id, n.name, p.module, p.build_id, p.repository, p.source_path, p.since, p.state, p.note,
p.settled_at, p.raised_at, p.acknowledged_at, p.cleared_at`
// 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 scanPending(rows pgx.Rows) ([]PendingAssignment, error) {
defer rows.Close()
var out []PendingAssignment
for rows.Next() {
var p PendingAssignment
if err := rows.Scan(&p.ID, &p.Node, &p.Module, &p.Build, &p.Repository, &p.Path, &p.Since, &p.State,
&p.Note, &p.Settled, &p.Raised, &p.Acknowledged, &p.Cleared); err != nil {
return nil, err
}
out = append(out, p)
}
return out, rows.Err()
func pendingOf(v chain.Assignment) PendingAssignment {
p := v.Pending
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}
}
// RecordPending keeps an assignment until its module is registered. One is open per machine and module.
// 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 {
if keep(*k.Value.Pending) {
out = append(out, pendingOf(k.Value))
}
}
return out
}
// changePending changes the pending assignment of an id 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(v *chain.Assignment) 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]
for k := range held {
if held[k].Value.Pending.ID == id {
at = &held[k]
}
}
if at == nil {
return false, nil
}
v := at.Value
p := *v.Pending
v.Pending = &p
if !change(&v) {
return false, nil
}
if _, err := c.Assignments.Update(ctx, at.Key, v, 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
}
err = i.store.Pool().QueryRow(ctx,
`insert into pending_assignment (node, module, build_id, repository, source_path)
values ($1, $2, $3, $4, $5)
on conflict (node, module) where state in ('waiting', 'applying') do nothing
returning id, since`,
node.ID, p.Module, p.Build, p.Repository, strings.Trim(p.Path, "/")).Scan(&p.ID, &p.Since)
if errors.Is(err, pgx.ErrNoRows) {
return PendingAssignment{}, ErrAlreadyPending
}
c, err := i.theChain(ctx)
if err != nil {
return PendingAssignment{}, err
}
p.State = PendingWaiting
return p, nil
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}}
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 int64, build, repository, sourcePath string) (bool, error) {
tag, err := i.store.Pool().Exec(ctx,
`update pending_assignment set build_id = $2, repository = $3, source_path = $4
where id = $1 and state = 'waiting'`, id, build, repository, strings.Trim(sourcePath, "/"))
if err != nil {
return false, err
}
return tag.RowsAffected() == 1, nil
func (i *Inventory) RepointPending(ctx context.Context, id string, build, repository, sourcePath string) (bool, error) {
return i.changePending(ctx, id, func(v *chain.Assignment) bool {
if v.Pending.State != PendingWaiting {
return false
}
v.Pending.Build, v.Pending.Repository, v.Pending.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) {
if endedSince.IsZero() {
rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+`
from pending_assignment p join node n on n.id = p.node
where p.state = 'waiting' order by p.since, p.id`)
if err != nil {
return nil, err
}
return scanPending(rows)
}
rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+`
from pending_assignment p join node n on n.id = p.node
where p.state in ('waiting', 'applying') or p.settled_at >= $1
order by p.since, p.id`, endedSince)
held, err := i.pendingValues(ctx)
if err != nil {
return nil, err
}
return scanPending(rows)
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, newest first.
// PendingFor is the pending assignment of one module to one machine: the one its key holds now, which is the newest
// (an earlier one is read from the key's history, which nothing here needs).
func (i *Inventory) PendingFor(ctx context.Context, node, module string) ([]PendingAssignment, error) {
rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+`
from pending_assignment p join node n on n.id = p.node
where n.name = $1 and p.module = $2 order by p.since desc, p.id desc`, node, module)
held, err := i.pendingValues(ctx)
if err != nil {
return nil, err
}
return scanPending(rows)
var out []PendingAssignment
for _, k := range held {
if k.Value.Node == node && k.Value.Module == module {
out = append(out, pendingOf(k.Value))
}
}
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) {
rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+`
from pending_assignment p join node n on n.id = p.node
where p.state = 'applying' and p.settled_at < $1 order by p.id`, before)
held, err := i.pendingValues(ctx)
if err != nil {
return nil, err
}
return scanPending(rows)
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) {
rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+`
from pending_assignment p join node n on n.id = p.node
where p.state in ('expired', 'refused') and p.raised_at is null order by p.id`)
held, err := i.pendingValues(ctx)
if err != nil {
return nil, err
}
return scanPending(rows)
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) {
rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+`
from pending_assignment p join node n on n.id = p.node
where p.raised_at is not null and p.cleared_at is null order by p.id`)
held, err := i.pendingValues(ctx)
if err != nil {
return nil, err
}
return scanPending(rows)
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 int64, what string) error {
column := map[string]string{"raised": "raised_at", "acknowledged": "acknowledged_at", "cleared": "cleared_at"}[what]
if column == "" {
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")
}
_, err := i.store.Pool().Exec(ctx,
`update pending_assignment set `+column+` = coalesce(`+column+`, now()) where id = $1`, id)
now := time.Now().UTC()
_, err := i.changePending(ctx, id, func(v *chain.Assignment) bool {
mark := map[string]**time.Time{"raised": &v.Pending.Raised, "acknowledged": &v.Pending.Acknowledged, "cleared": &v.Pending.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 int64) (bool, error) {
tag, err := i.store.Pool().Exec(ctx,
`update pending_assignment set state = 'applying', settled_at = now() where id = $1 and state = 'waiting'`, id)
if err != nil {
return false, err
}
return tag.RowsAffected() == 1, nil
func (i *Inventory) ClaimPending(ctx context.Context, id string) (bool, error) {
now := time.Now().UTC()
return i.changePending(ctx, id, func(v *chain.Assignment) bool {
if !chain.CanMove(v.Pending.State, PendingApplying) {
return false
}
v.Pending.State, v.Pending.Settled = PendingApplying, &now
return true
})
}
// ReleaseClaim puts a claimed pending assignment back to waiting.
func (i *Inventory) ReleaseClaim(ctx context.Context, id int64) error {
_, err := i.store.Pool().Exec(ctx,
`update pending_assignment set state = 'waiting', settled_at = null where id = $1 and state = 'applying'`, id)
func (i *Inventory) ReleaseClaim(ctx context.Context, id string) error {
_, err := i.changePending(ctx, id, func(v *chain.Assignment) bool {
if v.Pending.State != PendingApplying {
return false
}
v.Pending.State, v.Pending.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 int64, from, state, note string) (bool, error) {
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")
}
tag, err := i.store.Pool().Exec(ctx,
`update pending_assignment set state = $3, note = $4, settled_at = now()
where id = $1 and state = $2`, id, from, state, note)
if err != nil {
return false, err
}
return tag.RowsAffected() == 1, nil
now := time.Now().UTC()
return i.changePending(ctx, id, func(v *chain.Assignment) bool {
if v.Pending.State != from || !chain.CanSettle(from, state) {
return false
}
v.Pending.State, v.Pending.Note, v.Pending.Settled = state, note, &now
return true
})
}
// ForgetEndedPending deletes ended pending assignments settled more than KeptFor ago, and says how many. One
// raised as a condition is kept until that condition is cleared: deleting it would leave the condition
// with nothing to answer it.
func (i *Inventory) ForgetEndedPending(ctx context.Context, now time.Time) (int64, error) {
tag, err := i.store.Pool().Exec(ctx,
`delete from pending_assignment where state not in ('waiting', 'applying') and settled_at < $1
and (raised_at is null or cleared_at is not null)`,
now.Add(-KeptFor))
if err != nil {
return 0, err
}
return tag.RowsAffected(), nil
}
// PendingKnown says which of these pending assignments are still on record.
func (i *Inventory) PendingKnown(ctx context.Context, ids []int64) (map[int64]bool, error) {
rows, err := i.store.Pool().Query(ctx, `select id from pending_assignment where id = any($1)`, ids)
// PendingKnown says which of these pending assignments are still on record: a key holds them now. One a newer pending
// assignment of its key replaced is not.
func (i *Inventory) PendingKnown(ctx context.Context, ids []string) (map[string]bool, error) {
held, err := i.pendingValues(ctx)
if err != nil {
return nil, err
}
defer rows.Close()
out := map[int64]bool{}
for rows.Next() {
var id int64
if err := rows.Scan(&id); err != nil {
return nil, err
}
out[id] = true
want := map[string]bool{}
for _, id := range ids {
want[id] = true
}
return out, rows.Err()
out := map[string]bool{}
for _, k := range held {
if want[k.Value.Pending.ID] {
out[k.Value.Pending.ID] = true
}
}
return out, nil
}