Merge main (#227) into the grants step's branch
This commit is contained in:
@@ -0,0 +1,72 @@
|
||||
-- A declaration says the assignment generation it was composed from, and every send is recorded
|
||||
-- (novox/hq issue 234).
|
||||
--
|
||||
-- On 2026-10-04 a declaration newer in sequence than every other named four fewer modules than the
|
||||
-- assignments held, and the machine undeclared all four. The sequence orders arrival; it cannot tell a
|
||||
-- later send that carries an older view of the assignments. And nothing the mesh kept said which process
|
||||
-- sent it, from what view: the node row holds only the digest of the last send.
|
||||
|
||||
-- The assignment generation: one counter for the mesh, raised in the same transaction as every change to
|
||||
-- what is assigned where. A trigger, not a line in each writer: an assignment is also taken by a node's
|
||||
-- removal (on delete cascade) and renamed with its module (on update cascade), which no writer in Go
|
||||
-- sees, and a rule kept only by the writers that remember it is the rule this issue found broken. Seeded
|
||||
-- at 1, so every declaration composed after this claims a generation and none claims zero ("none").
|
||||
create table assignment_generation (
|
||||
one boolean primary key default true check (one),
|
||||
generation bigint not null
|
||||
);
|
||||
insert into assignment_generation (one, generation) values (true, 1);
|
||||
|
||||
create function raise_assignment_generation() returns trigger language plpgsql as $$
|
||||
begin
|
||||
update assignment_generation set generation = generation + 1;
|
||||
return null;
|
||||
end
|
||||
$$;
|
||||
|
||||
-- Per row, so a statement that changes nothing raises nothing: an assignment repeated (`on conflict do
|
||||
-- nothing` inserts no row) and an update that leaves a row as it was (the WHEN below; a plain UPDATE fires a
|
||||
-- row trigger whether or not it changed anything). A statement that changes several rows raises once per
|
||||
-- row, which only ever moves it forward. Two triggers, because only an update has both OLD and NEW.
|
||||
create trigger assignment_generation_raised
|
||||
after insert or delete on assignment
|
||||
for each row execute function raise_assignment_generation();
|
||||
create trigger assignment_generation_raised_by_update
|
||||
after update on assignment
|
||||
for each row when (old is distinct from new) execute function raise_assignment_generation();
|
||||
|
||||
-- The generation a machine was last sent, written in the same transaction as the digest of that send, so
|
||||
-- what the mesh WOULD send is stamped the same way and reads as byte for byte what it DID send when
|
||||
-- nothing else changed (as sent_epoch, migration 0068). Null for a send without one.
|
||||
alter table node add column sent_generation bigint;
|
||||
-- Whether the machine's node-engine said it reads a generation (its reports' `reads_generation`): until
|
||||
-- it has, it is sent none, because an older node-engine refuses a key it does not know, whole.
|
||||
alter table node add column reads_generation boolean not null default false;
|
||||
|
||||
-- Every send: the machine, its sequence, who sent it — the lease epoch the sender acted under and the
|
||||
-- caller, process and build — the generation it was composed from, its digest and the modules it named.
|
||||
-- Written after the declaration is away, with the node row's record of it. A refusal by the machine of a
|
||||
-- send for its generation is kept on the send it refused, so the condition it raises names the sender
|
||||
-- from here; a refusal of a sequence the mesh has no send for is kept as a row of its own, `recorded`
|
||||
-- false, whose sender is said to be unknown. `counter_raised_to` is the generation the mesh's counter was
|
||||
-- raised to when the refusal showed the counter behind the machine (a store put back from a backup). The
|
||||
-- newest 200 per machine are kept.
|
||||
create table declaration_send (
|
||||
id bigserial primary key,
|
||||
node uuid not null references node (id) on delete cascade,
|
||||
sequence bigint,
|
||||
epoch bigint,
|
||||
sender text not null,
|
||||
generation bigint,
|
||||
digest text not null,
|
||||
modules text[] not null default '{}',
|
||||
sent_at timestamptz not null default now(),
|
||||
refused_at timestamptz,
|
||||
-- refused_applied is the generation the machine said it had applied when it refused this send.
|
||||
refused_applied bigint,
|
||||
counter_raised_to bigint,
|
||||
recorded boolean not null default true
|
||||
);
|
||||
create index declaration_send_node on declaration_send (node, id desc);
|
||||
create index declaration_send_sequence on declaration_send (node, sequence);
|
||||
create index declaration_send_refused on declaration_send (refused_at) where refused_at is not null;
|
||||
@@ -0,0 +1,287 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// The assignment generation and the record of every send (novox/hq issue 234, migration 0094).
|
||||
//
|
||||
// A declaration carries the generation of the assignments it was composed from, and a machine refuses one
|
||||
// composed from an older generation than it applied — the sequence orders arrival and cannot tell a later
|
||||
// send carrying an older view. And every send is written down with who sent it, from which generation and
|
||||
// naming which modules, so what a machine was last told, by whom, is read back rather than guessed at.
|
||||
|
||||
// sendsKept is how many sends are kept per machine: the newest. Enough to read back a morning of pushes
|
||||
// on a busy machine; the record is for finding who sent what, not an archive.
|
||||
const sendsKept = 200
|
||||
|
||||
// AssignmentGeneration is the mesh's assignment generation now: raised by the store itself in the same
|
||||
// transaction as every change to what is assigned where (a trigger on the assignment table).
|
||||
func (i *Inventory) AssignmentGeneration(ctx context.Context) (int64, error) {
|
||||
var g int64
|
||||
if err := i.store.Pool().QueryRow(ctx, `select generation from assignment_generation`).Scan(&g); err != nil {
|
||||
return 0, fmt.Errorf("reading the assignment generation: %w", err)
|
||||
}
|
||||
return g, nil
|
||||
}
|
||||
|
||||
// RaiseAssignmentGeneration raises the generation past one a machine applied, and answers it: one more
|
||||
// than that, or what it already was when it is past it. It never lowers it.
|
||||
//
|
||||
// For one case only: a machine refused a send composed from the generation the mesh holds now, so the
|
||||
// machine applied a higher one than the mesh's counter — a store put back from a backup. Every send after
|
||||
// would be refused for ever; raised past it, the next send carries a generation the machine takes, composed
|
||||
// from the assignments the store holds now. A stale sender never meets this: its generation is below the
|
||||
// counter, and the counter is not touched for it.
|
||||
func (i *Inventory) RaiseAssignmentGeneration(ctx context.Context, past int64) (int64, error) {
|
||||
var g int64
|
||||
err := i.store.Pool().QueryRow(ctx,
|
||||
`update assignment_generation set generation = greatest(generation, $1 + 1) returning generation`, past).Scan(&g)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("raising the assignment generation past %d: %w", past, err)
|
||||
}
|
||||
return g, nil
|
||||
}
|
||||
|
||||
// SentGeneration is the generation a machine was last sent, by its id; zero for one sent without. What the
|
||||
// mesh WOULD send is stamped with it, so it reads as byte for byte what it DID send when nothing else changed.
|
||||
func (i *Inventory) SentGeneration(ctx context.Context, id string) (int64, error) {
|
||||
var g *int64
|
||||
if err := i.store.Pool().QueryRow(ctx, `select sent_generation from node where id = $1`, id).Scan(&g); err != nil {
|
||||
return 0, fmt.Errorf("reading the generation %s was last sent: %w", id, err)
|
||||
}
|
||||
if g == nil {
|
||||
return 0, nil
|
||||
}
|
||||
return *g, nil
|
||||
}
|
||||
|
||||
// ReadsGeneration says a machine's node-engine said it reads a generation in a declaration, by its id.
|
||||
func (i *Inventory) ReadsGeneration(ctx context.Context, id string) (bool, error) {
|
||||
var reads bool
|
||||
if err := i.store.Pool().QueryRow(ctx, `select reads_generation from node where id = $1`, id).Scan(&reads); err != nil {
|
||||
return false, fmt.Errorf("reading whether %s reads a generation: %w", id, err)
|
||||
}
|
||||
return reads, nil
|
||||
}
|
||||
|
||||
// RecordReadsGeneration keeps what a machine's latest report said of reading a generation.
|
||||
func (i *Inventory) RecordReadsGeneration(ctx context.Context, id string, reads bool) error {
|
||||
_, err := i.store.Pool().Exec(ctx, `update node set reads_generation = $2 where id = $1`, id, reads)
|
||||
return err
|
||||
}
|
||||
|
||||
// Send is one declaration sent to a machine, as the mesh records it.
|
||||
type Send struct {
|
||||
// Node is the machine's id; NodeName its name, filled where it is read back.
|
||||
Node string
|
||||
NodeName string
|
||||
Sequence int64
|
||||
// Epoch is the lease epoch its sender acted under, zero for one that acted under none.
|
||||
Epoch int64
|
||||
// Sender is who sent it, in words: the caller, and the process and build that composed it.
|
||||
Sender string
|
||||
// Generation is the assignment generation it was composed from, zero when not known.
|
||||
Generation int64
|
||||
Digest string
|
||||
// Modules are the modules it named.
|
||||
Modules []string
|
||||
SentAt time.Time
|
||||
// RefusedAt is when the machine refused it for its generation, nil when it did not; RefusedApplied the
|
||||
// generation the machine said it had applied then; CounterRaisedTo what the mesh's counter was raised to
|
||||
// because the refusal showed it behind the machine, zero when it was not.
|
||||
RefusedAt *time.Time
|
||||
RefusedApplied int64
|
||||
CounterRaisedTo int64
|
||||
// Recorded is false for a refusal of a sequence the mesh has no send for: its sender is unknown.
|
||||
Recorded bool
|
||||
}
|
||||
|
||||
// unknownSender is the sender of a refused sequence the mesh has no record of sending.
|
||||
const unknownSender = "a sender the mesh has no record of (no send of this sequence was recorded: sent by " +
|
||||
"hand, before sends were recorded, or by a controller whose record was not written)"
|
||||
|
||||
// SendNotRecordedError is a send whose machine's record was written — its digest and the generation it
|
||||
// carried — and whose record of who sent it was not. The send is away and the machine's record stands; the
|
||||
// caller says this loudly and does not take the send for failed.
|
||||
type SendNotRecordedError struct{ Err error }
|
||||
|
||||
func (e *SendNotRecordedError) Error() string {
|
||||
return "the record of the send was not written: " + e.Err.Error()
|
||||
}
|
||||
func (e *SendNotRecordedError) Unwrap() error { return e.Err }
|
||||
|
||||
// RecordSentWith writes down what a machine was just sent — its digest, the builds it carried, the epoch and
|
||||
// the generation on the wire — and the send itself, who sent it and from which generation (novox/hq issue
|
||||
// 234), in one transaction. The digest and the generation are written together, so the would-send is never
|
||||
// stamped with a generation the machine was not sent. The send's own record is written under a savepoint: if
|
||||
// it cannot be, the machine's record is still committed and a *SendNotRecordedError says so. The oldest
|
||||
// sends beyond the newest 200 for the machine are let go.
|
||||
func (i *Inventory) RecordSentWith(ctx context.Context, node, digest string, builds map[string]string, epoch uint64,
|
||||
generation int64, s Send) error {
|
||||
var sentEpoch *int64
|
||||
if epoch > 0 {
|
||||
e := int64(epoch)
|
||||
sentEpoch = &e
|
||||
}
|
||||
var carried *string
|
||||
if builds != nil {
|
||||
raw, err := json.Marshal(builds)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
text := string(raw)
|
||||
carried = &text
|
||||
}
|
||||
tx, err := i.store.Pool().Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
if _, err := tx.Exec(ctx,
|
||||
`update node set sent = $2, sent_at = now(), sent_builds = $3::jsonb, sent_epoch = $4, sent_generation = $5
|
||||
where id = $1`, node, digest, carried, sentEpoch, nullIfZero(generation)); err != nil {
|
||||
return err
|
||||
}
|
||||
recordErr := recordSend(ctx, tx, node, s)
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if recordErr != nil {
|
||||
return &SendNotRecordedError{Err: recordErr}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// recordSend writes one send under a savepoint of tx, which it rolls back when the send cannot be written.
|
||||
func recordSend(ctx context.Context, tx pgx.Tx, node string, s Send) error {
|
||||
sp, err := tx.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = sp.Rollback(ctx) }()
|
||||
modules := s.Modules
|
||||
if modules == nil {
|
||||
modules = []string{}
|
||||
}
|
||||
if _, err := sp.Exec(ctx,
|
||||
`insert into declaration_send (node, sequence, epoch, sender, generation, digest, modules)
|
||||
values ($1, $2, $3, $4, $5, $6, $7)`,
|
||||
node, nullIfZero(s.Sequence), nullIfZero(s.Epoch), s.Sender, nullIfZero(s.Generation), s.Digest,
|
||||
modules); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := pruneSends(ctx, sp, node); err != nil {
|
||||
return err
|
||||
}
|
||||
return sp.Commit(ctx)
|
||||
}
|
||||
|
||||
// pruneSends lets go of a machine's sends beyond the newest 200.
|
||||
func pruneSends(ctx context.Context, q queries, node string) error {
|
||||
_, err := q.Exec(ctx,
|
||||
`delete from declaration_send where node = $1 and id not in
|
||||
(select id from declaration_send where node = $1 order by id desc limit $2)`, node, sendsKept)
|
||||
return err
|
||||
}
|
||||
|
||||
// sendColumns are a send's columns as scanSends reads them.
|
||||
const sendColumns = `s.node, n.name, coalesce(s.sequence, 0), coalesce(s.epoch, 0), s.sender, coalesce(s.generation, 0),
|
||||
s.digest, s.modules, s.sent_at, s.refused_at, coalesce(s.refused_applied, 0), coalesce(s.counter_raised_to, 0),
|
||||
s.recorded`
|
||||
|
||||
func scanSends(rows pgx.Rows) ([]Send, error) {
|
||||
defer rows.Close()
|
||||
var out []Send
|
||||
for rows.Next() {
|
||||
var s Send
|
||||
if err := rows.Scan(&s.Node, &s.NodeName, &s.Sequence, &s.Epoch, &s.Sender, &s.Generation, &s.Digest,
|
||||
&s.Modules, &s.SentAt, &s.RefusedAt, &s.RefusedApplied, &s.CounterRaisedTo, &s.Recorded); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, s)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// LastSend is the last declaration a machine was sent, by its name; false when none was recorded. A refusal
|
||||
// of a sequence the mesh has no send for is not a send, and is not it.
|
||||
func (i *Inventory) LastSend(ctx context.Context, name string) (Send, bool, error) {
|
||||
rows, err := i.store.Pool().Query(ctx, `select `+sendColumns+`
|
||||
from declaration_send s join node n on n.id = s.node
|
||||
where n.name = $1 and s.recorded order by s.id desc limit 1`, name)
|
||||
if err != nil {
|
||||
return Send{}, false, err
|
||||
}
|
||||
sends, err := scanSends(rows)
|
||||
if err != nil || len(sends) == 0 {
|
||||
return Send{}, false, err
|
||||
}
|
||||
return sends[0], true, nil
|
||||
}
|
||||
|
||||
// RefusedSend keeps a machine's refusal of the send of a sequence for the generation it came from, and
|
||||
// answers that send, sender and all; false when no send of that sequence was recorded (RecordUnrecordedRefusal
|
||||
// keeps that one). The newest of that sequence when there are several: a declaration a person sent by hand may
|
||||
// repeat one. raisedTo is what the mesh's counter was raised to for it, zero for none.
|
||||
func (i *Inventory) RefusedSend(ctx context.Context, node string, sequence, applied, raisedTo int64) (Send, bool, error) {
|
||||
rows, err := i.store.Pool().Query(ctx, `with refused as (
|
||||
update declaration_send set refused_at = now(), refused_applied = $3, counter_raised_to = $4
|
||||
where id = (select id from declaration_send where node = $1 and sequence = $2 and recorded
|
||||
order by id desc limit 1)
|
||||
returning *)
|
||||
select `+sendColumns+` from refused s join node n on n.id = s.node`, node, sequence, applied,
|
||||
nullIfZero(raisedTo))
|
||||
if err != nil {
|
||||
return Send{}, false, err
|
||||
}
|
||||
sends, err := scanSends(rows)
|
||||
if err != nil || len(sends) == 0 {
|
||||
return Send{}, false, err
|
||||
}
|
||||
return sends[0], true, nil
|
||||
}
|
||||
|
||||
// RecordUnrecordedRefusal keeps a machine's refusal of a sequence the mesh has no send for, as a row of its
|
||||
// own whose sender is unknown, and answers it: S20 raises it like any other, because a refusal nobody can
|
||||
// attribute is no quieter for that. s carries the machine, the sequence, epoch, generation and digest the
|
||||
// refused declaration carried, and the refusal's applied generation and counter raise.
|
||||
func (i *Inventory) RecordUnrecordedRefusal(ctx context.Context, s Send) (Send, error) {
|
||||
rows, err := i.store.Pool().Query(ctx, `with refused as (
|
||||
insert into declaration_send (node, sequence, epoch, sender, generation, digest, refused_at, refused_applied,
|
||||
counter_raised_to, recorded)
|
||||
values ($1, $2, $3, $4, $5, $6, now(), $7, $8, false) returning *)
|
||||
select `+sendColumns+` from refused s join node n on n.id = s.node`,
|
||||
s.Node, nullIfZero(s.Sequence), nullIfZero(s.Epoch), unknownSender, nullIfZero(s.Generation), s.Digest,
|
||||
s.RefusedApplied, nullIfZero(s.CounterRaisedTo))
|
||||
if err != nil {
|
||||
return Send{}, err
|
||||
}
|
||||
sends, err := scanSends(rows)
|
||||
if err != nil {
|
||||
return Send{}, err
|
||||
}
|
||||
if len(sends) == 0 {
|
||||
return Send{}, fmt.Errorf("the refusal of sequence %d was not kept", s.Sequence)
|
||||
}
|
||||
if err := pruneSends(ctx, i.store.Pool(), s.Node); err != nil {
|
||||
return Send{}, err
|
||||
}
|
||||
return sends[0], nil
|
||||
}
|
||||
|
||||
// RefusedSendsSince is every send refused for its generation since a moment, newest first.
|
||||
func (i *Inventory) RefusedSendsSince(ctx context.Context, since time.Time) ([]Send, error) {
|
||||
rows, err := i.store.Pool().Query(ctx, `select `+sendColumns+`
|
||||
from declaration_send s join node n on n.id = s.node
|
||||
where s.refused_at >= $1 order by s.refused_at desc`, since)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return scanSends(rows)
|
||||
}
|
||||
@@ -0,0 +1,207 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The assignment generation and the record of every send (novox/hq issue 234). On 2026-10-04 a declaration
|
||||
// newer in sequence than every other named four fewer modules than the assignments held, a machine applied
|
||||
// it, and nothing the mesh kept could say who sent it or from what view.
|
||||
|
||||
func generationNow(t *testing.T, inv *Inventory) int64 {
|
||||
t.Helper()
|
||||
g, err := inv.AssignmentGeneration(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// **Every change to what is assigned where raises the generation, in its own transaction** — an
|
||||
// assignment, an unassignment, and one taken by the machine's removal, which no writer in Go makes — and
|
||||
// a repeated assignment, which changes nothing, does not.
|
||||
func TestEveryAssignmentChangeRaisesTheGeneration(t *testing.T) {
|
||||
inv, node := aNodeWithModules(t, "postgres", "web")
|
||||
ctx := t.Context()
|
||||
start := generationNow(t, inv)
|
||||
if start < 1 {
|
||||
t.Fatalf("the generation starts at %d; zero is \"none claimed\" on the wire", start)
|
||||
}
|
||||
if _, err := inv.Assign(ctx, node, "postgres"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assigned := generationNow(t, inv)
|
||||
if assigned <= start {
|
||||
t.Fatalf("an assignment left the generation at %d", assigned)
|
||||
}
|
||||
if _, err := inv.Assign(ctx, node, "postgres"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if again := generationNow(t, inv); again != assigned {
|
||||
t.Errorf("an assignment repeated, which changed nothing, moved the generation from %d to %d", assigned, again)
|
||||
}
|
||||
if err := inv.Unassign(ctx, node, "postgres"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
unassigned := generationNow(t, inv)
|
||||
if unassigned <= assigned {
|
||||
t.Fatalf("an unassignment left the generation at %d", unassigned)
|
||||
}
|
||||
if _, err := inv.Assign(ctx, node, "web"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := generationNow(t, inv)
|
||||
if _, err := inv.RemoveNodeForTest(ctx, node); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if removed := generationNow(t, inv); removed <= before {
|
||||
t.Errorf("a machine's removal took its assignments and left the generation at %d", removed)
|
||||
}
|
||||
}
|
||||
|
||||
// **The generation never goes down, and is raised past what a machine applied when the mesh's own counter
|
||||
// is behind it** — a store put back from a backup — so the next send is not refused for ever.
|
||||
func TestTheGenerationIsRaisedPastWhatAMachineApplied(t *testing.T) {
|
||||
inv, _ := aNodeWithModules(t)
|
||||
now := generationNow(t, inv)
|
||||
if raised, err := inv.RaiseAssignmentGeneration(t.Context(), now+40); err != nil || raised != now+41 {
|
||||
t.Fatalf("raised past %d to %d (%v)", now+40, raised, err)
|
||||
}
|
||||
if raised, err := inv.RaiseAssignmentGeneration(t.Context(), 2); err != nil || raised != now+41 {
|
||||
t.Fatalf("a raise below the counter moved it to %d (%v)", raised, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **Every send is recorded: its sequence, its sender, the generation it came from and the modules it
|
||||
// named** — and the machine's last send is what `plan` reads, the would-send is stamped with the generation
|
||||
// written with the digest, and a refusal is kept on the send it refused, naming its sender.
|
||||
func TestASendIsRecordedWithItsSenderGenerationAndModules(t *testing.T) {
|
||||
inv, node := aNodeWithModules(t)
|
||||
ctx := t.Context()
|
||||
record, err := inv.NodeByName(ctx, node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
first := Send{Sequence: 11, Epoch: 57, Sender: "the controller's daemon (pid 7 on anchor)",
|
||||
Generation: 40, Digest: "d11", Modules: []string{"docker", "pacman", "sudo"}}
|
||||
if err := inv.RecordSentWith(ctx, record.ID, "d11", nil, 57, 40, first); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stale := Send{Sequence: 12, Epoch: 57, Sender: "a one-shot push by jochen at a shell on anchor",
|
||||
Generation: 38, Digest: "d12", Modules: []string{"docker"}}
|
||||
if err := inv.RecordSentWith(ctx, record.ID, "d12", nil, 57, 38, stale); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
last, found, err := inv.LastSend(ctx, node)
|
||||
if err != nil || !found {
|
||||
t.Fatalf("the last send is not kept: %v", err)
|
||||
}
|
||||
if last.Sequence != 12 || last.Sender != stale.Sender || last.Generation != 38 || !last.Recorded ||
|
||||
!slices.Equal(last.Modules, []string{"docker"}) || last.SentAt.IsZero() {
|
||||
t.Fatalf("the last send reads %+v", last)
|
||||
}
|
||||
if generation, err := inv.SentGeneration(ctx, record.ID); err != nil || generation != 38 {
|
||||
t.Fatalf("what the machine was last sent of it reads %d (%v)", generation, err)
|
||||
}
|
||||
|
||||
refused, found, err := inv.RefusedSend(ctx, record.ID, 12, 40, 0)
|
||||
if err != nil || !found || refused.Sender != stale.Sender || refused.RefusedApplied != 40 ||
|
||||
refused.CounterRaisedTo != 0 {
|
||||
t.Fatalf("the refusal is not kept on the send it refused: %+v, %v, %v", refused, found, err)
|
||||
}
|
||||
since, err := inv.RefusedSendsSince(ctx, time.Now().Add(-time.Hour))
|
||||
if err != nil || len(since) != 1 || since[0].NodeName != node || since[0].Sequence != 12 {
|
||||
t.Fatalf("the refused sends within the hour read %+v (%v)", since, err)
|
||||
}
|
||||
if _, found, err := inv.RefusedSend(ctx, record.ID, 99, 40, 0); err != nil || found {
|
||||
t.Errorf("a refusal of a send never recorded was found: %v, %v", found, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A refusal of a sequence the mesh has no send for is kept too, its sender unknown**, so S20 raises it;
|
||||
// it is not the machine's last send, and it says what the counter was raised to.
|
||||
func TestARefusalOfASendNeverRecordedIsKeptWithAnUnknownSender(t *testing.T) {
|
||||
inv, node := aNodeWithModules(t)
|
||||
ctx := t.Context()
|
||||
record, err := inv.NodeByName(ctx, node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
kept, err := inv.RecordUnrecordedRefusal(ctx, Send{Node: record.ID, Sequence: 99, Epoch: 57, Generation: 41,
|
||||
Digest: "d99", RefusedApplied: 44, CounterRaisedTo: 45})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if kept.Recorded || kept.Sender != unknownSender || kept.RefusedAt == nil || kept.CounterRaisedTo != 45 ||
|
||||
kept.NodeName != node {
|
||||
t.Fatalf("the refusal reads %+v", kept)
|
||||
}
|
||||
since, err := inv.RefusedSendsSince(ctx, time.Now().Add(-time.Hour))
|
||||
if err != nil || len(since) != 1 || since[0].Sequence != 99 {
|
||||
t.Fatalf("the refused sends within the hour read %+v (%v)", since, err)
|
||||
}
|
||||
if _, found, err := inv.LastSend(ctx, node); err != nil || found {
|
||||
t.Errorf("a refusal of a send never recorded reads as the machine's last send: %v, %v", found, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **The machine's record and the send's are written together, and a send record that cannot be written
|
||||
// leaves the machine's record standing and says so**: the digest and generation are the machine's, and a
|
||||
// send that is away must not read as failed, nor the machine as behind.
|
||||
func TestASendRecordThatCannotBeWrittenLeavesTheMachinesRecord(t *testing.T) {
|
||||
inv, node := aNodeWithModules(t)
|
||||
ctx := t.Context()
|
||||
record, err := inv.NodeByName(ctx, node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// A NUL byte is text PostgreSQL refuses: the send's own row cannot be written.
|
||||
broken := Send{Sequence: 13, Sender: "a test", Generation: 42, Digest: "d\x0013"}
|
||||
err = inv.RecordSentWith(ctx, record.ID, "d13", nil, 0, 42, broken)
|
||||
var unrecorded *SendNotRecordedError
|
||||
if !errors.As(err, &unrecorded) {
|
||||
t.Fatalf("a send record that could not be written read as %v", err)
|
||||
}
|
||||
if generation, err := inv.SentGeneration(ctx, record.ID); err != nil || generation != 42 {
|
||||
t.Fatalf("the machine's generation was not written with its digest: %d (%v)", generation, err)
|
||||
}
|
||||
if sent, err := inv.Outstanding(ctx, node); err != nil || sent != "d13" {
|
||||
t.Fatalf("the machine's digest was not written: %q (%v)", sent, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **An update that leaves an assignment as it was raises nothing** (the trigger's WHEN).
|
||||
func TestAnUpdateThatChangesNoAssignmentRaisesNothing(t *testing.T) {
|
||||
inv, node := aNodeWithModules(t, "web")
|
||||
ctx := t.Context()
|
||||
if _, err := inv.Assign(ctx, node, "web"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := generationNow(t, inv)
|
||||
if _, err := inv.store.Pool().Exec(ctx, `update assignment set module = module`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if after := generationNow(t, inv); after != before {
|
||||
t.Errorf("an update that changed nothing moved the generation from %d to %d", before, after)
|
||||
}
|
||||
}
|
||||
|
||||
// **A machine that reads a generation is recorded from its reports**, and one rolled back says so no longer.
|
||||
func TestWhetherAMachineReadsAGenerationIsItsLatestWord(t *testing.T) {
|
||||
inv, node := aNodeWithModules(t)
|
||||
record, err := inv.NodeByName(t.Context(), node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, reads := range []bool{false, true, false} {
|
||||
if err := inv.RecordReadsGeneration(t.Context(), record.ID, reads); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, err := inv.ReadsGeneration(t.Context(), record.ID); err != nil || got != reads {
|
||||
t.Fatalf("recorded %v, read %v (%v)", reads, got, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user