Roll a core build back by a witness that is not the new build (hq to-be 45 Phase 4)
The launcher trusted a counter only a by-hand reconcile ever cleared and a known-good nothing in the daemon wrote, so no machine could roll its host back; the controller and the node tools were replaced in place with nothing kept. - The launcher runs a delivered host that is not known-good on trial: one that crashes, stops for nothing, or does not report within ten minutes goes back to known-good, once per version, recorded in rolled-back. The host proves itself when the mesh takes a report under its own build, says every standing verdict on its reports, never stands aside for a rolled-back version, and restarts its service once when its launcher was replaced on disk. - The engine keeps the controller's and the node tools' previous build beside the new one and judges the new one: the lease taken by the controller it started (read-only direct get of mesh-controller_lease/holder), or this machine's runtime answering $SRV.PING.node-tools.<node>, within sixty seconds of time it could ask. Not healthy: the previous restored, once, said. Proved: the previous deleted. A build declared not-reversible is never rolled back. - Retire never removes a version newer than the running one.
This commit is contained in:
+81
-12
@@ -8,9 +8,11 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-host/internal/declaration"
|
||||
"github.com/novox/mesh-host/internal/store"
|
||||
"github.com/novox/mesh-host/internal/witness"
|
||||
)
|
||||
|
||||
// Running the mesh's own code, without the module choosing how.
|
||||
@@ -105,20 +107,50 @@ func applyProcess(ctx context.Context, r *declaration.Process, run Runner,
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Replaced rather than merged: the bundle is the whole of what it runs, and files left from a
|
||||
// previous version would be loaded by a runtime that walks a directory.
|
||||
if err := os.RemoveAll(at); err != nil {
|
||||
return out, err
|
||||
// **A core process keeps the build before it until the new one is proved** (novox/hq to-be 45
|
||||
// §8): the engine's witness judges the new build and restores the kept one when it is not healthy
|
||||
// in bound. Placing decides what happens to the directory first — the running build moved aside,
|
||||
// a build on trial discarded, a build rolled back here refused, the running build kept when it is
|
||||
// the one declared — and keeps that, so a restoration later knows which build is where.
|
||||
by := witnessOf(r)
|
||||
placing := witness.Placing{Unpack: true}
|
||||
if by == witness.ByNone {
|
||||
// Nothing judged. Whatever was kept for a process that was judged before goes.
|
||||
if err := witness.Forget(daemonRoot, r.Name); err != nil {
|
||||
return out, err
|
||||
}
|
||||
// Replaced rather than merged: the bundle is the whole of what it runs, and files left from a
|
||||
// previous version would be loaded by a runtime that walks a directory.
|
||||
if err := os.RemoveAll(at); err != nil {
|
||||
return out, err
|
||||
}
|
||||
} else {
|
||||
placing, err = witness.Place(daemonRoot, r.Name, by, got, digestWritten(previous.Wrote), r.NotReversible, time.Now())
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
}
|
||||
if err := os.MkdirAll(at, 0o755); err != nil {
|
||||
return out, err
|
||||
written := 0
|
||||
if placing.Unpack {
|
||||
if err := os.MkdirAll(at, 0o755); err != nil {
|
||||
return out, err
|
||||
}
|
||||
if written, err = unpack(body, at); err != nil {
|
||||
return out, err
|
||||
}
|
||||
if err := ownAll(at, r.User); err != nil {
|
||||
return out, err
|
||||
}
|
||||
}
|
||||
written, err := unpack(body, at)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
if err := ownAll(at, r.User); err != nil {
|
||||
return out, err
|
||||
if placing.Commit != nil {
|
||||
// Kept whatever the start below does: a build that would not start is still the build placed,
|
||||
// and it is the witness's to judge.
|
||||
defer func() {
|
||||
if err := placing.Commit(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "mesh-host: placed %s, and what the witness keeps about it could not be saved: %v\n",
|
||||
r.Name, err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// **A step is run to completion, not installed.** What follows it is gated on it finishing,
|
||||
@@ -201,9 +233,15 @@ func applyProcess(ctx context.Context, r *declaration.Process, run Runner,
|
||||
out.Action = "created"
|
||||
}
|
||||
out.Detail = fmt.Sprintf("%d file(s), running as %s.service", written, r.Name)
|
||||
if !placing.Unpack {
|
||||
out.Detail = fmt.Sprintf("its build already in place, running as %s.service", r.Name)
|
||||
}
|
||||
if because != "" {
|
||||
out.Detail += ", restarted because " + because + " changed"
|
||||
}
|
||||
if placing.Detail != "" {
|
||||
out.Detail += "; " + placing.Detail
|
||||
}
|
||||
out.wrote = want
|
||||
return out, nil
|
||||
}
|
||||
@@ -358,6 +396,13 @@ func removeProcess(ctx context.Context, a store.Applied, run Runner) (string, st
|
||||
}
|
||||
}
|
||||
bundle := filepath.Join(daemonRoot, name)
|
||||
// And the build kept before it, with what the witness knew (novox/hq to-be 45 §8).
|
||||
if _, err := os.Stat(witness.Dir(daemonRoot, name)); err == nil {
|
||||
found = true
|
||||
if err := witness.Forget(daemonRoot, name); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
}
|
||||
if _, err := os.Stat(bundle); err == nil {
|
||||
found = true
|
||||
if err := os.RemoveAll(bundle); err != nil {
|
||||
@@ -381,3 +426,27 @@ func runFrom(r *declaration.Process) []string {
|
||||
}
|
||||
return run
|
||||
}
|
||||
|
||||
// witnessOf is how a process's new builds are judged: as it declares, or by its name's default — the
|
||||
// mesh's two core processes (novox/hq to-be 45 §8). A step or a scheduled run is never judged.
|
||||
func witnessOf(r *declaration.Process) string {
|
||||
if r.RunOnce || r.Schedule != "" {
|
||||
return witness.ByNone
|
||||
}
|
||||
if r.Witness != "" {
|
||||
return r.Witness
|
||||
}
|
||||
return witness.Default(r.Name)
|
||||
}
|
||||
|
||||
// digestWritten is the bundle digest a process's record says was placed: the first half of what it
|
||||
// wrote, "sha256:<hex> <unit>".
|
||||
func digestWritten(wrote string) string {
|
||||
if fields := strings.Fields(wrote); len(fields) > 0 && strings.HasPrefix(fields[0], "sha256:") {
|
||||
return fields[0]
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// DaemonRoot is where unpacked daemons live — and what a witness keeps beside them.
|
||||
func DaemonRoot() string { return daemonRoot }
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
package apply
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-host/internal/store"
|
||||
"github.com/novox/mesh-host/internal/witness"
|
||||
)
|
||||
|
||||
// A core process keeps the build before it while the new one is judged (novox/hq to-be 45 §8): the
|
||||
// applier places the new build where the unit runs it from, and moves the old one aside rather than
|
||||
// deleting it. A process nobody judges is replaced as ever.
|
||||
func TestACoreProcessKeepsTheBuildBeforeItWhileTheNewOneIsJudged(t *testing.T) {
|
||||
units, bundles := t.TempDir(), t.TempDir()
|
||||
wasUnits, wasBundles := unitDir, daemonRoot
|
||||
unitDir, daemonRoot = units, bundles
|
||||
t.Cleanup(func() { unitDir, daemonRoot = wasUnits, wasBundles })
|
||||
run := func(ctx context.Context, name string, args ...string) (string, error) { return "", nil }
|
||||
|
||||
oldBody, oldDigest := anArchive(t, map[string]string{"node-tools": "old\n"})
|
||||
newBody, newDigest := anArchive(t, map[string]string{"node-tools": "new\n"})
|
||||
runtime := func(body []byte, digest string) string {
|
||||
return `{"id":"node-tools.runtime","type":"process","name":"node-tools","source":"` + serving(t, body) +
|
||||
`","digest":"` + digest + `","run":["./node-tools"]}`
|
||||
}
|
||||
|
||||
_, state, err := Apply(context.Background(), archHost(t), declare(t, runtime(oldBody, oldDigest)), store.State{},
|
||||
store.OriginDeclared, run, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(witness.Trials(bundles)) != 0 {
|
||||
t.Fatal("a first placement went on trial with nothing to go back to")
|
||||
}
|
||||
_, _, err = Apply(context.Background(), archHost(t), declare(t, runtime(newBody, newDigest)), state,
|
||||
store.OriginDeclared, run, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, _ := os.ReadFile(filepath.Join(bundles, "node-tools", "node-tools")); string(got) != "new\n" {
|
||||
t.Fatalf("the new build is not where the unit runs it from: %q", got)
|
||||
}
|
||||
if got, _ := os.ReadFile(filepath.Join(witness.Dir(bundles, "node-tools"), "previous", "node-tools")); string(got) != "old\n" {
|
||||
t.Fatalf("the build before was not kept beside it: %q", got)
|
||||
}
|
||||
trials := witness.Trials(bundles)
|
||||
if len(trials) != 1 || trials[0].Running != newDigest || trials[0].Previous != oldDigest || trials[0].Witness != witness.ByPing {
|
||||
t.Fatalf("the new build is not on trial against the old one: %+v", trials)
|
||||
}
|
||||
// Undeclared: everything kept about it goes with it.
|
||||
if _, _, err := removeProcess(context.Background(), store.Applied{Target: "node-tools"}, run); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := os.Stat(witness.Dir(bundles, "node-tools")); !os.IsNotExist(err) {
|
||||
t.Fatalf("what the witness kept outlived the process: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A build rolled back on this machine is not placed again while the mesh still declares it; the
|
||||
// restored build keeps running, and the apply says why.
|
||||
func TestARolledBackBuildIsNotPlacedAgain(t *testing.T) {
|
||||
units, bundles := t.TempDir(), t.TempDir()
|
||||
wasUnits, wasBundles := unitDir, daemonRoot
|
||||
unitDir, daemonRoot = units, bundles
|
||||
t.Cleanup(func() { unitDir, daemonRoot = wasUnits, wasBundles })
|
||||
run := func(ctx context.Context, name string, args ...string) (string, error) { return "", nil }
|
||||
|
||||
oldBody, oldDigest := anArchive(t, map[string]string{"mesh-controller": "old\n"})
|
||||
newBody, newDigest := anArchive(t, map[string]string{"mesh-controller": "new\n"})
|
||||
controller := func(body []byte, digest, env string) string {
|
||||
return `{"id":"mesh-controller.controller","type":"process","name":"mesh-controller","source":"` + serving(t, body) +
|
||||
`","digest":"` + digest + `","run":["./mesh-controller","serve"],"env":{"X":"` + env + `"}}`
|
||||
}
|
||||
_, state, err := Apply(context.Background(), archHost(t), declare(t, controller(oldBody, oldDigest, "1")), store.State{},
|
||||
store.OriginDeclared, run, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, state, err = Apply(context.Background(), archHost(t), declare(t, controller(newBody, newDigest, "1")), state,
|
||||
store.OriginDeclared, run, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := witness.Restore(context.Background(), bundles, "mesh-controller", "never took the lease", witness.Runner(run),
|
||||
time.Now()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The mesh still declares the new build, and its unit changes: the process is re-placed.
|
||||
report, _, err := Apply(context.Background(), archHost(t), declare(t, controller(newBody, newDigest, "2")), state,
|
||||
store.OriginDeclared, run, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, _ := os.ReadFile(filepath.Join(bundles, "mesh-controller", "mesh-controller")); string(got) != "old\n" {
|
||||
t.Fatalf("the rolled-back build was placed again: %q", got)
|
||||
}
|
||||
if o := outcomeOf(report, "mesh-controller.controller"); !strings.Contains(o.Detail, "rolled back on this machine") {
|
||||
t.Fatalf("the apply does not say why it kept the build before: %+v", o)
|
||||
}
|
||||
}
|
||||
@@ -590,6 +590,19 @@ type Process struct {
|
||||
// For a process that stays up; a step or a scheduled run is not running a moment later by
|
||||
// design, so there is nothing to hand over to.
|
||||
Replaces []string `json:"replaces,omitempty"`
|
||||
|
||||
// Witness is how the node-engine judges a new build of this process, and restores the build
|
||||
// before it when the new one is not healthy in bound (novox/hq to-be 45 §8): "lease" — the
|
||||
// controller this machine started holds the controller's lease; "ping" — this machine's runtime
|
||||
// answers the services protocol's PING; "none". Absent is the default for the process's name:
|
||||
// the mesh's two core processes are judged, nothing else is. For a process that stays up.
|
||||
Witness string `json:"witness,omitempty"`
|
||||
|
||||
// NotReversible says why this build may not be rolled back, when it may not: the build before it
|
||||
// would run against what this one changes — a migration it runs that the older build cannot read
|
||||
// (to-be 45 §8, rule 8). A build so declared that is not healthy in bound is left running and said
|
||||
// as urgent; the build before it is never started against the newer data.
|
||||
NotReversible string `json:"not-reversible,omitempty"`
|
||||
}
|
||||
|
||||
func (d *Process) Identity() string { return d.ID }
|
||||
@@ -637,6 +650,19 @@ func (d *Process) validate(where string, _ bool) []string {
|
||||
if len(d.Run) == 0 {
|
||||
problems = append(problems, where+": a process needs to say what to run")
|
||||
}
|
||||
switch d.Witness {
|
||||
case "", "lease", "ping", "none":
|
||||
default:
|
||||
problems = append(problems, fmt.Sprintf("%s: witness %q is not one this host keeps: lease, ping or none",
|
||||
where, d.Witness))
|
||||
}
|
||||
if d.Witness != "" && d.Witness != "none" && (d.RunOnce || d.Schedule != "") {
|
||||
problems = append(problems, where+": a witness judges a process that stays up; a step or a "+
|
||||
"scheduled run is not running between its runs")
|
||||
}
|
||||
if strings.ContainsAny(d.NotReversible, "\n\r") {
|
||||
problems = append(problems, where+": not-reversible is one line")
|
||||
}
|
||||
for _, part := range d.Run {
|
||||
if part == "" {
|
||||
problems = append(problems, where+": a process command has an empty element")
|
||||
|
||||
@@ -194,6 +194,69 @@ type Report struct {
|
||||
// Rekey is this node taking a found tunnel's key as its overlay key after enrolment (novox/hq
|
||||
// ADR 0105). Not an account of the machine: a report carrying one says nothing else.
|
||||
Rekey *Rekey `json:"rekey,omitempty"`
|
||||
|
||||
// Rollbacks is what this machine's witnesses decided about a core build that was not healthy
|
||||
// in bound (novox/hq to-be 45 §8, ADR 0227 rule 8): the node-engine's launcher about the
|
||||
// engine, the engine about the controller and the node tools. **Said on every report while it
|
||||
// stands** — while the build it judged is still the one the mesh asks this machine to run — so
|
||||
// a report lost, or a controller restarted, never makes a rollback silent. Empty once a newer
|
||||
// build has proved itself.
|
||||
Rollbacks []Rollback `json:"rollbacks,omitempty"`
|
||||
|
||||
// Witness is the version of the witness contract this node-engine keeps (to-be 45 §8): its
|
||||
// presence says it reads a process's `witness` and `not-reversible`, which a strict decoder
|
||||
// older than it refuses — so the controller sends either only to a machine whose report carries
|
||||
// it, as it sends `epoch` only where `report_sequence` was seen (ADR 0229).
|
||||
Witness int `json:"witness,omitempty"`
|
||||
}
|
||||
|
||||
// WitnessContract is the version of the witness contract this host keeps: the fields above, the
|
||||
// process fields `witness` and `not-reversible`, and what each witness reads (internal/witness).
|
||||
const WitnessContract = 1
|
||||
|
||||
// The core components a witness judges (to-be 45 §8), as a rollback names them.
|
||||
const (
|
||||
ComponentEngine = "node-engine"
|
||||
ComponentController = "controller"
|
||||
ComponentNodeTools = "node-tools"
|
||||
)
|
||||
|
||||
// What a witness concluded, as a rollback says it.
|
||||
const (
|
||||
// RolledBack is the previous build restored and running.
|
||||
RolledBack = "rolled-back"
|
||||
// NotReversible is a build that failed its health and was declared not reversible: the
|
||||
// previous build would run against what the new one already changed (a migration that ran), so
|
||||
// nothing was restored. The failing build is left as it is, and this is urgent.
|
||||
NotReversible = "not-reversible"
|
||||
// NothingToRestore is a build that failed its health with no previous build kept to go back to.
|
||||
NothingToRestore = "nothing-to-restore"
|
||||
// RestoreFailed is a restoration that was attempted and did not complete.
|
||||
RestoreFailed = "restore-failed"
|
||||
// Unwitnessed is a build whose health could not be judged at all within the bound — the
|
||||
// witness could not ask (no grant, no lease bucket) — so it is neither proved nor rolled back.
|
||||
Unwitnessed = "unwitnessed"
|
||||
// Halted is the node-engine's launcher giving up: the build it would go back to fails too, so
|
||||
// the fault is the machine's and not a build's.
|
||||
Halted = "halted"
|
||||
)
|
||||
|
||||
// Rollback is one witness's verdict on one core build (to-be 45 §8). The controller raises
|
||||
// `core.<component>.<node>.<outcome>` from it; RolledBack, NotReversible, RestoreFailed and Halted
|
||||
// are urgent.
|
||||
type Rollback struct {
|
||||
// Component is which core component: node-engine, controller or node-tools.
|
||||
Component string `json:"component"`
|
||||
// From is the build that was judged: a host version for the node-engine, the bundle's digest for
|
||||
// a process.
|
||||
From string `json:"from"`
|
||||
// To is the build restored, and empty when none was.
|
||||
To string `json:"to,omitempty"`
|
||||
// Outcome is one of the words above.
|
||||
Outcome string `json:"outcome"`
|
||||
// Why is what the witness saw, in words for a person.
|
||||
Why string `json:"why"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// Order is where a declaration stands among everything the mesh has sent this node (novox/hq to-be
|
||||
|
||||
@@ -128,6 +128,12 @@ func TestTheWireFormatIsExactlyTheseFieldNames(t *testing.T) {
|
||||
[]string{"id", "module", "kind", "target", "since", "changed", "kept"}},
|
||||
{Reach{Protocol: "tcp", Address: "0.0.0.0", Port: 8080, By: "c", Published: true, ContainerPort: 80},
|
||||
[]string{"protocol", "address", "port", "by", "published", "container-port"}},
|
||||
// novox/hq to-be 45 §8: a witness's verdicts, said on every report while they stand, and the
|
||||
// witness contract this host keeps.
|
||||
{Report{Node: "n", Rollbacks: []Rollback{{Component: ComponentController}}, Witness: WitnessContract},
|
||||
[]string{"node", "rollbacks", "witness"}},
|
||||
{Rollback{Component: ComponentNodeTools, From: "sha256:b", To: "sha256:a", Outcome: RolledBack, Why: "w"},
|
||||
[]string{"component", "from", "to", "outcome", "why", "at"}},
|
||||
} {
|
||||
raw, err := json.Marshal(c.value)
|
||||
if err != nil {
|
||||
|
||||
+29
@@ -0,0 +1,29 @@
|
||||
# A bus with the grants a witness needs (novox/hq to-be 45 §8), for witnessing_nats_test.go:
|
||||
#
|
||||
# docker run -d --rm --name w -p 14224:4222 -v $PWD/internal/link/testdata:/c:ro nats:2.11-alpine -js -c /c/witness-grants.conf
|
||||
# MESH_TEST_NATS_GRANTS=nats://127.0.0.1:14224 go test ./internal/link/ -run TestNatsWitness
|
||||
#
|
||||
# `node.anchor` is a machine's host with exactly the two subjects the witness asks added to what a
|
||||
# host already has (its inbox); `node.bare` is a host without them. `runtime` stands in for the tool
|
||||
# runtime, `admin` for the controller writing the lease.
|
||||
jetstream: enabled
|
||||
accounts {
|
||||
MESH {
|
||||
jetstream: enabled
|
||||
users = [
|
||||
{ user: "node.anchor", password: "a", permissions: {
|
||||
publish: { allow: ["$JS.API.DIRECT.GET.KV_mesh-controller_lease.$KV.mesh-controller_lease.holder", "$SRV.PING.node-tools.anchor"] }
|
||||
subscribe: { allow: ["_INBOX.node.anchor.>"] }
|
||||
} }
|
||||
{ user: "node.bare", password: "b", permissions: {
|
||||
publish: { allow: ["mesh.control.bare.>"] }
|
||||
subscribe: { allow: ["_INBOX.node.bare.>"] }
|
||||
} }
|
||||
{ user: "runtime", password: "r", permissions: {
|
||||
subscribe: { allow: ["$SRV.PING.node-tools.>"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
{ user: "admin", password: "x" }
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// What a witness asks the bus (novox/hq to-be 45 §8): the node-engine judging the controller and the
|
||||
// node tools it placed on this machine, by what the bus says about them rather than by what they
|
||||
// say about themselves.
|
||||
//
|
||||
// **Two questions, both read-only, both on the host's own link.** The controller's lease key, read
|
||||
// by JetStream's direct get — one subject, no stream information, nothing written; and this
|
||||
// machine's tool runtime, asked the NATS services protocol's PING by its name and this machine's —
|
||||
// so only this machine's runtime can answer it.
|
||||
|
||||
// ErrCannotAsk is a question the bus did not let this host put, or did not answer: no grant for it,
|
||||
// no such bucket, the bus slow. **It says nothing about the build being judged**, so a witness
|
||||
// counts none of it against the build's bound.
|
||||
var ErrCannotAsk = errors.New("this host cannot ask the bus")
|
||||
|
||||
// ErrNoAnswer is a question the bus delivered and nobody answered in time: the build being judged is
|
||||
// not there to answer. Counted against its bound.
|
||||
var ErrNoAnswer = errors.New("nothing answered")
|
||||
|
||||
// Asker is what a witness asks through. The link open now implements it; nil while there is none.
|
||||
type Asker interface {
|
||||
// ReadKey is a key-value bucket's current value for a key. Found false is nobody holding it —
|
||||
// absent, deleted or expired; an error wrapping ErrCannotAsk is no reading at all.
|
||||
ReadKey(ctx context.Context, bucket, key string) (value []byte, found bool, err error)
|
||||
// Ping asks the service `service`'s instance `id` whether it is there. Nil is an answer;
|
||||
// ErrNoAnswer is none in time; ErrCannotAsk is no asking.
|
||||
Ping(ctx context.Context, service, id string) error
|
||||
}
|
||||
|
||||
// DirectGetSubject is the one subject a host asks a bucket's key on: JetStream's direct get of the
|
||||
// bucket's stream, for the key's own subject. What a host's grant names exactly — no wildcard.
|
||||
func DirectGetSubject(bucket, key string) string {
|
||||
return "$JS.API.DIRECT.GET.KV_" + bucket + ".$KV." + bucket + "." + key
|
||||
}
|
||||
|
||||
// PingSubject is the services protocol's PING of one instance of one service.
|
||||
func PingSubject(service, id string) string { return "$SRV.PING." + service + "." + id }
|
||||
|
||||
// Asker is the link open now, when there is one and it can ask.
|
||||
func (q *Queue) Asker() Asker {
|
||||
q.init()
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
if a, ok := q.bus.(Asker); ok {
|
||||
return a
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *natsLink) ReadKey(ctx context.Context, bucket, key string) ([]byte, bool, error) {
|
||||
subject := DirectGetSubject(bucket, key)
|
||||
msg, err := l.conn.RequestWithContext(ctx, subject, nil)
|
||||
switch {
|
||||
case errors.Is(err, nats.ErrNoResponders):
|
||||
return nil, false, fmt.Errorf("%w: the bus has no bucket %s that answers a direct get", ErrCannotAsk, bucket)
|
||||
case err != nil:
|
||||
return nil, false, fmt.Errorf("%w: reading %s from %s: %v%s", ErrCannotAsk, key, bucket, err, l.refusal(subject))
|
||||
}
|
||||
if msg.Header != nil {
|
||||
switch status := msg.Header.Get("Status"); status {
|
||||
case "":
|
||||
case "404":
|
||||
return nil, false, nil
|
||||
default:
|
||||
return nil, false, fmt.Errorf("%w: reading %s from %s: the bus answered %s %s", ErrCannotAsk, key,
|
||||
bucket, status, msg.Header.Get("Description"))
|
||||
}
|
||||
switch msg.Header.Get("KV-Operation") {
|
||||
case "DEL", "PURGE":
|
||||
return nil, false, nil
|
||||
}
|
||||
}
|
||||
if len(msg.Data) == 0 {
|
||||
return nil, false, nil
|
||||
}
|
||||
return msg.Data, true, nil
|
||||
}
|
||||
|
||||
func (l *natsLink) Ping(ctx context.Context, service, id string) error {
|
||||
subject := PingSubject(service, id)
|
||||
msg, err := l.conn.RequestWithContext(ctx, subject, nil)
|
||||
switch {
|
||||
case errors.Is(err, nats.ErrNoResponders):
|
||||
// The bus delivered the question and nothing subscribes to it: the runtime is not on the bus.
|
||||
return fmt.Errorf("%w: no %s on this machine is on the bus", ErrNoAnswer, service)
|
||||
case err != nil:
|
||||
if refused := l.refusal(subject); refused != "" {
|
||||
return fmt.Errorf("%w: asking %s: %v%s", ErrCannotAsk, subject, err, refused)
|
||||
}
|
||||
return fmt.Errorf("%w: %s did not answer: %v", ErrNoAnswer, subject, err)
|
||||
}
|
||||
var ping struct {
|
||||
Name string `json:"name"`
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Name != service || ping.ID != id {
|
||||
return fmt.Errorf("%w: what answered %s is not %s %s", ErrNoAnswer, subject, service, id)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// refusal is the bus's last word on this connection when it refused a publish to subject, said as
|
||||
// the end of an error; empty when it did not.
|
||||
func (l *natsLink) refusal(subject string) string {
|
||||
last := l.conn.LastError()
|
||||
if last == nil {
|
||||
return ""
|
||||
}
|
||||
words := strings.ToLower(last.Error())
|
||||
if strings.Contains(words, "permissions violation") && strings.Contains(last.Error(), subject) {
|
||||
return " — the bus refused it: this host's grant does not name " + subject
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// What a witness asks, asked of a real server with the grants the mesh composes (novox/hq to-be 45 §8):
|
||||
// whether the direct get and the PING work as the two subjects named, and whether a host without them
|
||||
// is told apart — "cannot ask", never "the build did not answer". See testdata/witness-grants.conf.
|
||||
func TestNatsWitnessAsksWhatItIsGranted(t *testing.T) {
|
||||
url := os.Getenv("MESH_TEST_NATS_GRANTS")
|
||||
if url == "" {
|
||||
t.Skip("MESH_TEST_NATS_GRANTS unset")
|
||||
}
|
||||
dial := func(user, password, inbox string) *nats.Conn {
|
||||
t.Helper()
|
||||
conn, err := nats.Connect(url, nats.UserInfo(user, password), nats.CustomInboxPrefix(inbox))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(conn.Close)
|
||||
return conn
|
||||
}
|
||||
admin := dial("admin", "x", "_INBOX.admin")
|
||||
js, err := admin.JetStream()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = js.DeleteKeyValue("mesh-controller_lease")
|
||||
kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "mesh-controller_lease", TTL: 15 * time.Second, History: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
anchor := &natsLink{conn: dial("node.anchor", "a", "_INBOX.node.anchor")}
|
||||
bare := &natsLink{conn: dial("node.bare", "b", "_INBOX.node.bare")}
|
||||
ask := func() (context.Context, context.CancelFunc) {
|
||||
return context.WithTimeout(context.Background(), 2*time.Second)
|
||||
}
|
||||
|
||||
// Nobody holds it.
|
||||
ctx, cancel := ask()
|
||||
if _, found, err := anchor.ReadKey(ctx, "mesh-controller_lease", "holder"); err != nil || found {
|
||||
t.Fatalf("an empty lease read as %v, %v", found, err)
|
||||
}
|
||||
cancel()
|
||||
// Held.
|
||||
if _, err := kv.Put("holder", []byte(`{"instance":"controller@anchor pid 1 since now","epoch":3}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel = ask()
|
||||
value, found, err := anchor.ReadKey(ctx, "mesh-controller_lease", "holder")
|
||||
cancel()
|
||||
if err != nil || !found {
|
||||
t.Fatalf("a held lease read as %v, %v", found, err)
|
||||
}
|
||||
var holder struct {
|
||||
Instance string `json:"instance"`
|
||||
}
|
||||
if json.Unmarshal(value, &holder); holder.Instance == "" {
|
||||
t.Fatalf("the lease's value is %s", value)
|
||||
}
|
||||
// Given back.
|
||||
if err := kv.Delete("holder"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel = ask()
|
||||
if _, found, err := anchor.ReadKey(ctx, "mesh-controller_lease", "holder"); err != nil || found {
|
||||
t.Fatalf("a lease given back read as %v, %v", found, err)
|
||||
}
|
||||
cancel()
|
||||
// A host without the grant cannot ask — which is not "nobody holds it".
|
||||
ctx, cancel = ask()
|
||||
if _, _, err := bare.ReadKey(ctx, "mesh-controller_lease", "holder"); !errors.Is(err, ErrCannotAsk) {
|
||||
t.Fatalf("a host without the grant read the lease as %v, want it unable to ask", err)
|
||||
}
|
||||
cancel()
|
||||
|
||||
// No runtime on the bus: not there to answer.
|
||||
ctx, cancel = ask()
|
||||
if err := anchor.Ping(ctx, "node-tools", "anchor"); !errors.Is(err, ErrNoAnswer) {
|
||||
t.Fatalf("a PING nobody serves is %v, want no answer", err)
|
||||
}
|
||||
cancel()
|
||||
// The runtime, answering as the services protocol does.
|
||||
runtime := dial("runtime", "r", "_INBOX.runtime")
|
||||
sub, err := runtime.Subscribe("$SRV.PING.node-tools.anchor", func(m *nats.Msg) {
|
||||
_ = m.Respond([]byte(`{"type":"io.nats.micro.v1.ping_response","name":"node-tools","id":"anchor","version":"0.1.0"}`))
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
if err := runtime.Flush(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel = ask()
|
||||
if err := anchor.Ping(ctx, "node-tools", "anchor"); err != nil {
|
||||
t.Fatalf("this machine's runtime answered and the witness heard %v", err)
|
||||
}
|
||||
cancel()
|
||||
// And a host not granted the PING cannot ask, though the runtime is there.
|
||||
ctx, cancel = ask()
|
||||
if err := bare.Ping(ctx, "node-tools", "anchor"); !errors.Is(err, ErrCannotAsk) {
|
||||
t.Fatalf("a host without the grant heard %v, want it unable to ask", err)
|
||||
}
|
||||
cancel()
|
||||
|
||||
// No lease bucket at all — a controller from before the lease: cannot ask, not "nobody holds it".
|
||||
if err := js.DeleteKeyValue("mesh-controller_lease"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel = ask()
|
||||
if _, _, err := anchor.ReadKey(ctx, "mesh-controller_lease", "holder"); !errors.Is(err, ErrCannotAsk) {
|
||||
t.Fatalf("a bus with no lease bucket read as %v, want unable to ask", err)
|
||||
}
|
||||
cancel()
|
||||
}
|
||||
+33
-13
@@ -260,15 +260,29 @@ func Versions(dir string) ([]Delivered, error) {
|
||||
// Empty when the running version is the newest, which is the ordinary answer. The host asks this
|
||||
// between reconciles and nowhere else: standing aside mid-apply is the half-configured machine the
|
||||
// host exists to prevent (novox/hq ADR 0141).
|
||||
func Successor(dir, running string) (Delivered, bool, error) {
|
||||
//
|
||||
// **Never a version the launcher rolled back** (novox/hq to-be 45 §8): standing aside for one would
|
||||
// hand the launcher a version it refuses to start, and the host would stand aside after every apply
|
||||
// for ever. rolledBack is the versions the launcher's record names; nil is none.
|
||||
func Successor(dir, running string, rolledBack ...string) (Delivered, bool, error) {
|
||||
delivered, err := Versions(dir)
|
||||
if err != nil {
|
||||
return Delivered{}, false, err
|
||||
}
|
||||
if len(delivered) == 0 {
|
||||
refused := map[string]bool{}
|
||||
for _, v := range rolledBack {
|
||||
refused[v] = true
|
||||
}
|
||||
var candidates []Delivered
|
||||
for _, d := range delivered {
|
||||
if !refused[d.Version] {
|
||||
candidates = append(candidates, d)
|
||||
}
|
||||
}
|
||||
if len(candidates) == 0 {
|
||||
return Delivered{}, false, nil
|
||||
}
|
||||
newest := delivered[0]
|
||||
newest := candidates[0]
|
||||
// A machine whose running version is not among the delivered ones is the machine every mesh has
|
||||
// one of: the host was put there by hand before any of this existed. Treating that as "stand
|
||||
// aside" is correct — what was delivered is what the mesh asked for.
|
||||
@@ -295,19 +309,25 @@ func Retire(dir, running string) ([]string, error) {
|
||||
}
|
||||
|
||||
keep := map[string]bool{running: true}
|
||||
at := -1
|
||||
for i, d := range delivered {
|
||||
if d.Version != running {
|
||||
continue
|
||||
if d.Version == running {
|
||||
at = i
|
||||
break
|
||||
}
|
||||
// Its predecessor is the next one down the list, which is the next oldest.
|
||||
if i+1 < len(delivered) {
|
||||
keep[delivered[i+1].Version] = true
|
||||
}
|
||||
break
|
||||
}
|
||||
// A running version that was never delivered has no predecessor among these, so the newest
|
||||
// delivered one is what a rollback would reach for. Keep it.
|
||||
if len(keep) == 1 && len(delivered) > 0 {
|
||||
switch {
|
||||
case at >= 0:
|
||||
// **Newer than the running one is never retired** (novox/hq to-be 45 §8): it is a successor
|
||||
// delivered and not yet started, or one the launcher rolled back that the mesh still declares —
|
||||
// deleting either would have the next apply deliver it again. And its predecessor, the next one
|
||||
// down the list, is what a rollback starts.
|
||||
for i := 0; i <= at+1 && i < len(delivered); i++ {
|
||||
keep[delivered[i].Version] = true
|
||||
}
|
||||
case len(delivered) > 0:
|
||||
// A running version that was never delivered has no predecessor among these, so the newest
|
||||
// delivered one is what a rollback would reach for. Keep it.
|
||||
keep[delivered[0].Version] = true
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
package upgrade
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The host's own successor, witnessed by its launcher (novox/hq to-be 45 §8, ADR 0227 rule 8).
|
||||
//
|
||||
// The launcher runs a delivered host that is not the known-good one **on trial**, and rolls back
|
||||
// from one that crashes repeatedly, stops for nothing, or does not report within its bound. Two
|
||||
// files are the whole conversation between them, both plain enough for a shell:
|
||||
//
|
||||
// - known-good — written by this host when the mesh has taken a report it made about its
|
||||
// declaration under its own build. That is what ends a trial.
|
||||
// - rolled-back — written by the launcher: one line per verdict, tab-separated — from, to, when
|
||||
// (seconds since the epoch), outcome, why. Read here and said on every report while it stands.
|
||||
|
||||
// RolledBackName is the launcher's record of its verdicts, beside the state.
|
||||
const RolledBackName = "rolled-back"
|
||||
|
||||
// RolledBackPath is where it lives, given where the state lives.
|
||||
func RolledBackPath(statePath string) string {
|
||||
return filepath.Join(filepath.Dir(statePath), RolledBackName)
|
||||
}
|
||||
|
||||
// Verdict is one line of the launcher's record.
|
||||
type Verdict struct {
|
||||
From, To, Outcome, Why string
|
||||
At time.Time
|
||||
}
|
||||
|
||||
// ReadRolledBack is the launcher's record, oldest first. A line it cannot read is skipped and named
|
||||
// in the error, never guessed at; absence is no verdicts.
|
||||
func ReadRolledBack(path string) ([]Verdict, error) {
|
||||
file, err := os.Open(path)
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer file.Close()
|
||||
var out []Verdict
|
||||
var bad []string
|
||||
lines := bufio.NewScanner(file)
|
||||
for n := 1; lines.Scan(); n++ {
|
||||
line := strings.TrimRight(lines.Text(), "\r")
|
||||
if strings.TrimSpace(line) == "" {
|
||||
continue
|
||||
}
|
||||
fields := strings.SplitN(line, "\t", 5)
|
||||
if len(fields) != 5 || fields[0] == "" {
|
||||
bad = append(bad, strconv.Itoa(n))
|
||||
continue
|
||||
}
|
||||
seconds, err := strconv.ParseInt(fields[2], 10, 64)
|
||||
if err != nil {
|
||||
bad = append(bad, strconv.Itoa(n))
|
||||
continue
|
||||
}
|
||||
out = append(out, Verdict{From: fields[0], To: fields[1], At: time.Unix(seconds, 0).UTC(),
|
||||
Outcome: fields[3], Why: fields[4]})
|
||||
}
|
||||
if err := lines.Err(); err != nil {
|
||||
return out, err
|
||||
}
|
||||
if len(bad) > 0 {
|
||||
return out, fmt.Errorf("the launcher's record %s has line(s) %s that are not from, to, when, outcome, why",
|
||||
path, strings.Join(bad, ", "))
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// RolledBackVersions are the versions the launcher refuses to start again.
|
||||
func RolledBackVersions(verdicts []Verdict) []string {
|
||||
var out []string
|
||||
for _, v := range verdicts {
|
||||
out = append(out, v.From)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Proved is this host's version seen to report its declaration under its own build: the evidence a
|
||||
// trial waits for, and the only evidence known-good has ever claimed (novox/hq ADR 0005).
|
||||
//
|
||||
// - known-good becomes this version, which ends the launcher's trial;
|
||||
// - the start counter is cleared, so months of ordinary restarts never add up to a rollback;
|
||||
// - versions older than this one's predecessor are retired — never one newer, never this one;
|
||||
// - and when this version is not the one a rollback went back to, the launcher's verdicts end: a
|
||||
// newer host has proved itself, and what was concluded about an older one no longer stands.
|
||||
//
|
||||
// Returns the versions retired. Every step is attempted; the errors are joined.
|
||||
func Proved(statePath, versionsDir, version string) ([]string, error) {
|
||||
if strings.TrimSpace(version) == "" {
|
||||
return nil, errors.New("a host that does not know its own version cannot prove it")
|
||||
}
|
||||
var errs []error
|
||||
if err := RecordKnownGood(KnownGoodPath(statePath), version); err != nil {
|
||||
errs = append(errs, fmt.Errorf("recording %s as known-good: %w", version, err))
|
||||
}
|
||||
if err := ClearAttempts(AttemptsPath(statePath)); err != nil {
|
||||
errs = append(errs, fmt.Errorf("clearing the start counter: %w", err))
|
||||
}
|
||||
retired, err := Retire(versionsDir, version)
|
||||
if err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
verdicts, err := ReadRolledBack(RolledBackPath(statePath))
|
||||
if err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
wentBackTo := false
|
||||
for _, v := range verdicts {
|
||||
if v.To == version {
|
||||
wentBackTo = true
|
||||
}
|
||||
}
|
||||
if len(verdicts) > 0 && !wentBackTo {
|
||||
if err := os.Remove(RolledBackPath(statePath)); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
return retired, errors.Join(errs...)
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
package upgrade
|
||||
|
||||
import (
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The launcher's record, read as the launcher writes it: the launcher itself writes it here, so the
|
||||
// two cannot drift (novox/hq to-be 45 §8).
|
||||
func TestTheLaunchersRecordIsReadAsTheLauncherWritesIt(t *testing.T) {
|
||||
state := t.TempDir()
|
||||
libexec := t.TempDir()
|
||||
versions := filepath.Join(libexec, VersionsDirName)
|
||||
base := time.Now().Add(-2 * time.Hour)
|
||||
for i, v := range []string{"1.0", "2.0"} {
|
||||
binary := deliver(t, versions, v, base.Add(time.Duration(i)*time.Hour))
|
||||
// Each fails, so the one run iteration ends.
|
||||
if err := os.WriteFile(binary, []byte("#!/bin/sh\nexit 1\n"), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(state, KnownGoodName), []byte("1.0\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(state, AttemptsName), []byte("3\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
launch := exec.Command("sh", "../../packaging/nox-mesh-host-launch")
|
||||
launch.Env = append(os.Environ(), "MESH_HOST_STATE_DIR="+state, "MESH_HOST_LIBEXEC="+libexec,
|
||||
"MESH_HOST_BIN=/nonexistent", "MESH_HOST_RUN_ONCE=1", "MESH_HOST_BACKOFF=0")
|
||||
_ = launch.Run()
|
||||
verdicts, err := ReadRolledBack(RolledBackPath(filepath.Join(state, "state.json")))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(verdicts) == 0 {
|
||||
t.Fatal("the launcher rolled back and nothing was read from its record")
|
||||
}
|
||||
v := verdicts[0]
|
||||
if v.From != "2.0" || v.To != "1.0" || v.Outcome != "rolled-back" || !strings.Contains(v.Why, "failed 3 times") ||
|
||||
time.Since(v.At) > time.Minute {
|
||||
t.Fatalf("the record reads as %+v", v)
|
||||
}
|
||||
}
|
||||
|
||||
// A version the launcher rolled back is never what this host stands aside for: it would be refused,
|
||||
// and the host would stand aside after every apply for ever.
|
||||
func TestTheHostNeverStandsAsideForARolledBackVersion(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
base := time.Now().Add(-2 * time.Hour)
|
||||
deliver(t, dir, "1.0", base)
|
||||
deliver(t, dir, "2.0", base.Add(time.Hour))
|
||||
if _, waiting, err := Successor(dir, "1.0", "2.0"); err != nil || waiting {
|
||||
t.Fatalf("standing aside for a rolled-back version (%v, %v)", waiting, err)
|
||||
}
|
||||
deliver(t, dir, "3.0", base.Add(90*time.Minute))
|
||||
if next, waiting, err := Successor(dir, "1.0", "2.0"); err != nil || !waiting || next.Version != "3.0" {
|
||||
t.Fatalf("a newer version after a rollback is not stood aside for: %+v %v %v", next, waiting, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Nothing newer than the running version is retired: a successor waiting, or one rolled back that the
|
||||
// mesh still declares — either would only be delivered again.
|
||||
func TestRetireNeverRemovesANewerVersion(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
base := time.Now().Add(-4 * time.Hour)
|
||||
for i, v := range []string{"one", "two", "three", "four"} {
|
||||
deliver(t, dir, v, base.Add(time.Duration(i)*time.Hour))
|
||||
}
|
||||
removed, err := Retire(dir, "two")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(removed) != 0 {
|
||||
t.Fatalf("retired %v while running two: one is its predecessor, three and four are newer", removed)
|
||||
}
|
||||
}
|
||||
|
||||
// Proved: known-good, the counter cleared, older versions retired, and the launcher's verdicts ended —
|
||||
// unless this is the version a rollback went back to, about which they still stand.
|
||||
func TestProvingAHostEndsTheLaunchersVerdictsOnlyWhenItIsNewer(t *testing.T) {
|
||||
state := filepath.Join(t.TempDir(), "state.json")
|
||||
dir := t.TempDir()
|
||||
base := time.Now().Add(-4 * time.Hour)
|
||||
for i, v := range []string{"0.9", "1.0", "2.0", "3.0"} {
|
||||
deliver(t, dir, v, base.Add(time.Duration(i)*time.Hour))
|
||||
}
|
||||
record := "2.0\t1.0\t1759744800\trolled-back\tit failed 3 times in a row\n"
|
||||
if err := os.WriteFile(RolledBackPath(state), []byte(record), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(AttemptsPath(state), []byte("2\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// The version gone back to reports: it is known-good, and the rollback still stands.
|
||||
if _, err := Proved(state, dir, "1.0"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if v, _ := ReadRolledBack(RolledBackPath(state)); len(v) != 1 {
|
||||
t.Fatal("proving the version a rollback went back to ended the rollback")
|
||||
}
|
||||
if got, _ := ReadKnownGood(KnownGoodPath(state)); got != "1.0" {
|
||||
t.Fatalf("known-good is %q", got)
|
||||
}
|
||||
if raw, _ := os.ReadFile(AttemptsPath(state)); strings.TrimSpace(string(raw)) != "0" {
|
||||
t.Fatalf("the start counter was not cleared: %q", raw)
|
||||
}
|
||||
|
||||
// A newer one reports: the rollback ends, and what is older than its predecessor goes.
|
||||
retired, err := Proved(state, dir, "3.0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if v, _ := ReadRolledBack(RolledBackPath(state)); len(v) != 0 {
|
||||
t.Fatalf("a newer host proved itself and the old rollback still stands: %+v", v)
|
||||
}
|
||||
if !reflect.DeepEqual(retired, []string{"0.9", "1.0"}) {
|
||||
t.Fatalf("retired %v, want 0.9 and 1.0 — 2.0 is 3.0's predecessor", retired)
|
||||
}
|
||||
}
|
||||
|
||||
// A line the launcher's record cannot be read by is named, never guessed at.
|
||||
func TestAnUnreadableRecordLineIsNamed(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), RolledBackName)
|
||||
if err := os.WriteFile(path, []byte("2.0\t1.0\t1759744800\trolled-back\twhy\ngarbage\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
verdicts, err := ReadRolledBack(path)
|
||||
if len(verdicts) != 1 || err == nil || !strings.Contains(err.Error(), "line(s) 2") {
|
||||
t.Fatalf("read %+v, %v", verdicts, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,155 @@
|
||||
// Package witness is the node-engine watching a core build it placed — the controller on the control
|
||||
// node, the node tools on every machine — and restoring the build before it when the new one is not
|
||||
// healthy in bound (novox/hq to-be 45 §8, ADR 0227 rule 8: the component being replaced is never the
|
||||
// only witness of its successor).
|
||||
//
|
||||
// **The contract is this file.** What the engine reads to call a build healthy, from where, in what
|
||||
// shape, within which bound — said once, here, and held by contract_test.go. The controller's side
|
||||
// writes what is read here (mesh-controller internal/lease Holder); a change on either side is a
|
||||
// change to this file and its test.
|
||||
package witness
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
)
|
||||
|
||||
// The two witnesses, as a process declares which watches it (`witness`). A process that says none is
|
||||
// placed as ever; one that says nothing is judged by its name's default (Default).
|
||||
const (
|
||||
// ByLease: healthy when the controller this machine started holds the controller's lease.
|
||||
ByLease = "lease"
|
||||
// ByPing: healthy when this machine's runtime answers the services protocol's PING.
|
||||
ByPing = "ping"
|
||||
// ByNone: not judged.
|
||||
ByNone = "none"
|
||||
)
|
||||
|
||||
// The controller's lease, as the host reads it (novox/hq ADR 0229): the key `holder` in the bucket
|
||||
// `mesh-controller_lease`, whose keys live fifteen seconds unless renewed, renewed every five.
|
||||
const (
|
||||
LeaseBucket = "mesh-controller_lease"
|
||||
LeaseKey = "holder"
|
||||
// LeaseAge is the bucket's age for its key. A holder whose last renewal is older than this is
|
||||
// not holding it, whatever the key still says.
|
||||
LeaseAge = 15 * time.Second
|
||||
// Skew is how far the controller's clock and this host's may disagree about when it took the
|
||||
// lease. One machine, one clock, in practice: a margin, not a tolerance.
|
||||
Skew = 2 * time.Second
|
||||
)
|
||||
|
||||
// Bounds (to-be 45 §8). A build is given Within of time the witness could ask in; asked every Every.
|
||||
const (
|
||||
// ControllerWithin: the controller holds the lease within sixty seconds of starting.
|
||||
ControllerWithin = 60 * time.Second
|
||||
// NodeToolsWithin: the runtime is announced and answering within sixty seconds of starting,
|
||||
// each PING answered within PingWithin.
|
||||
NodeToolsWithin = 60 * time.Second
|
||||
PingWithin = 5 * time.Second
|
||||
Every = 5 * time.Second
|
||||
// GiveUp is how long a witness goes on when it cannot ask at all before it says the build is
|
||||
// unwitnessed: the grant missing, the bucket missing, the bus away the whole time.
|
||||
GiveUp = 30 * time.Minute
|
||||
)
|
||||
|
||||
// ControllerLease is the lease's value as the controller writes it — mesh-controller internal/lease
|
||||
// Holder, field for field by its JSON names. Only what the witness reads is required; a field the
|
||||
// controller adds later is ignored here.
|
||||
type ControllerLease struct {
|
||||
// Instance names one controller process: "controller@<machine> pid <pid> since <RFC 3339>".
|
||||
Instance string `json:"instance"`
|
||||
// Host is the machine it runs on, as its own hostname says.
|
||||
Host string `json:"host,omitempty"`
|
||||
// Build is the controller's build as it knows it; not read here — the toolchain stamps no
|
||||
// version, so it says "development build" — and a bundle's identity is its digest, which the
|
||||
// host already knows.
|
||||
Build string `json:"build,omitempty"`
|
||||
Epoch uint64 `json:"epoch,omitempty"`
|
||||
// Taken is when this instance took the key, Renewed when it last renewed it, by its own clock.
|
||||
Taken time.Time `json:"taken"`
|
||||
Renewed time.Time `json:"renewed"`
|
||||
}
|
||||
|
||||
// ParseLease reads the key's value.
|
||||
func ParseLease(value []byte) (ControllerLease, error) {
|
||||
var l ControllerLease
|
||||
if err := json.Unmarshal(value, &l); err != nil {
|
||||
return ControllerLease{}, fmt.Errorf("the lease's holder is not the controller's lease value: %w", err)
|
||||
}
|
||||
return l, nil
|
||||
}
|
||||
|
||||
// HeldBySince says whether the lease is held by a controller on machine `host` that took it at or
|
||||
// after `since` — the controller this machine started then, and not the one before it — and is
|
||||
// held now. Why says what it saw when it is not.
|
||||
func (l ControllerLease) HeldBySince(host string, since, now time.Time) (bool, string) {
|
||||
switch {
|
||||
case l.Instance == "":
|
||||
return false, "the lease names no holder"
|
||||
case host != "" && !sameMachine(l.Host, host):
|
||||
return false, fmt.Sprintf("the lease is held by %s, on %s and not this machine", l.Instance, l.Host)
|
||||
case l.Taken.Before(since.Add(-Skew)):
|
||||
return false, fmt.Sprintf("the lease is held by %s, taken %s — before the new build started at %s",
|
||||
l.Instance, l.Taken.UTC().Format(time.RFC3339), since.UTC().Format(time.RFC3339))
|
||||
case now.Sub(latest(l.Taken, l.Renewed)) > LeaseAge:
|
||||
return false, fmt.Sprintf("the lease names %s and was last renewed %s ago, past its %s",
|
||||
l.Instance, now.Sub(latest(l.Taken, l.Renewed)).Round(time.Second), LeaseAge)
|
||||
}
|
||||
return true, fmt.Sprintf("%s holds the lease, epoch %d", l.Instance, l.Epoch)
|
||||
}
|
||||
|
||||
// sameMachine compares two hostnames as names, so a short name and its fully qualified form agree.
|
||||
func sameMachine(a, b string) bool {
|
||||
short := func(s string) string {
|
||||
s = strings.ToLower(strings.TrimSpace(s))
|
||||
if i := strings.IndexByte(s, '.'); i > 0 {
|
||||
s = s[:i]
|
||||
}
|
||||
return s
|
||||
}
|
||||
return short(a) == short(b)
|
||||
}
|
||||
|
||||
func latest(a, b time.Time) time.Time {
|
||||
if b.After(a) {
|
||||
return b
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
// NodeToolsService is the runtime's name on the services protocol, and its instance is this
|
||||
// machine's node name: what `$SRV.PING.<service>.<node>` asks, so only this machine's runtime can
|
||||
// answer (mesh-tools node-tools internal/runtime, announce.Service{Name: module, ID: node}).
|
||||
const NodeToolsService = "node-tools"
|
||||
|
||||
// Default is the witness a process is judged by when it names none: the two core processes the mesh
|
||||
// composes, by the names it composes them under, and nothing else.
|
||||
func Default(process string) string {
|
||||
switch process {
|
||||
case "mesh-controller":
|
||||
return ByLease
|
||||
case NodeToolsService:
|
||||
return ByPing
|
||||
}
|
||||
return ByNone
|
||||
}
|
||||
|
||||
// Within is the bound for a witness.
|
||||
func Within(by string) time.Duration {
|
||||
if by == ByLease {
|
||||
return ControllerWithin
|
||||
}
|
||||
return NodeToolsWithin
|
||||
}
|
||||
|
||||
// Component names what a witness judges, as a rollback says it.
|
||||
func Component(by string) string {
|
||||
if by == ByLease {
|
||||
return link.ComponentController
|
||||
}
|
||||
return link.ComponentNodeTools
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
package witness
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
)
|
||||
|
||||
// The contract, held. Each of these is a line the controller's side relies on or writes; a change on
|
||||
// either side changes this test (novox/hq to-be 45 §8).
|
||||
|
||||
// The lease is read where the controller keeps it (ADR 0229), on the one subject a host's grant names.
|
||||
func TestTheLeaseIsReadWhereTheControllerKeepsIt(t *testing.T) {
|
||||
if LeaseBucket != "mesh-controller_lease" || LeaseKey != "holder" {
|
||||
t.Fatalf("the lease is read from %s/%s; the controller keeps it in mesh-controller_lease/holder",
|
||||
LeaseBucket, LeaseKey)
|
||||
}
|
||||
if got := link.DirectGetSubject(LeaseBucket, LeaseKey); got != "$JS.API.DIRECT.GET.KV_mesh-controller_lease.$KV.mesh-controller_lease.holder" {
|
||||
t.Fatalf("the host asks %s for the lease; its grant names exactly the direct get of the key", got)
|
||||
}
|
||||
if LeaseAge != 15*time.Second {
|
||||
t.Fatalf("the lease's age is %s; the controller's bucket keeps a key fifteen seconds", LeaseAge)
|
||||
}
|
||||
}
|
||||
|
||||
// What the controller writes — mesh-controller internal/lease Holder, by its JSON names — is what is
|
||||
// read. The value below is the shape that Holder marshals to.
|
||||
func TestTheLeaseValueIsTheControllersHolder(t *testing.T) {
|
||||
written := `{"instance":"controller@anchor pid 4242 since 2026-10-06T10:00:00Z","host":"anchor",` +
|
||||
`"build":"development build","epoch":57,"taken":"2026-10-06T10:00:01Z","renewed":"2026-10-06T10:00:31Z"}`
|
||||
l, err := ParseLease([]byte(written))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if l.Instance == "" || l.Host != "anchor" || l.Epoch != 57 || l.Taken.IsZero() || l.Renewed.IsZero() {
|
||||
t.Fatalf("the lease was not read whole: %+v", l)
|
||||
}
|
||||
// And a field the controller adds later does not stop it being read.
|
||||
if _, err := ParseLease([]byte(`{"instance":"i","taken":"2026-10-06T10:00:01Z","renewed":"2026-10-06T10:00:01Z","bundle":"sha256:x"}`)); err != nil {
|
||||
t.Fatalf("a lease value with a field this host does not know was refused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Healthy is: held now, on this machine, by an instance that took it after the new build started.
|
||||
func TestTheNewControllerHoldsTheLeaseOnlyWhenItTookItAfterItStarted(t *testing.T) {
|
||||
started := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
|
||||
now := started.Add(30 * time.Second)
|
||||
held := ControllerLease{Instance: "controller@anchor pid 2 since …", Host: "anchor.example",
|
||||
Taken: started.Add(3 * time.Second), Renewed: now.Add(-2 * time.Second), Epoch: 58}
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
l ControllerLease
|
||||
want bool
|
||||
says string
|
||||
}{
|
||||
{"held by the new instance", held, true, "holds the lease"},
|
||||
{"nobody", ControllerLease{}, false, "no holder"},
|
||||
{"the old instance, taken before the restart", func() ControllerLease {
|
||||
l := held
|
||||
l.Taken = started.Add(-time.Hour)
|
||||
return l
|
||||
}(), false, "before the new build started"},
|
||||
{"another machine's controller", func() ControllerLease {
|
||||
l := held
|
||||
l.Host = "home-server"
|
||||
return l
|
||||
}(), false, "not this machine"},
|
||||
{"taken and no longer renewed", func() ControllerLease {
|
||||
l := held
|
||||
l.Renewed = now.Add(-20 * time.Second)
|
||||
return l
|
||||
}(), false, "past its"},
|
||||
{"taken within the clocks' skew of the start", func() ControllerLease {
|
||||
l := held
|
||||
l.Taken = started.Add(-time.Second)
|
||||
return l
|
||||
}(), true, "holds the lease"},
|
||||
} {
|
||||
got, why := c.l.HeldBySince("anchor", started, now)
|
||||
if got != c.want || !strings.Contains(why, c.says) {
|
||||
t.Errorf("%s: healthy %v (%s), want %v saying %q", c.name, got, why, c.want, c.says)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The runtime is asked by its own name and this machine's: only this machine's runtime can answer.
|
||||
func TestTheRuntimeIsAskedByItsNameAndThisMachines(t *testing.T) {
|
||||
if got := link.PingSubject(NodeToolsService, "anchor"); got != "$SRV.PING.node-tools.anchor" {
|
||||
t.Fatalf("the host asks %s; the runtime answers $SRV.PING.node-tools.<node>", got)
|
||||
}
|
||||
}
|
||||
|
||||
// What is judged by default is the mesh's two core processes, by the names the controller composes
|
||||
// them under, and nothing else.
|
||||
func TestOnlyTheCoreProcessesAreJudgedByDefault(t *testing.T) {
|
||||
for name, want := range map[string]string{"mesh-controller": ByLease, "node-tools": ByPing, "greeter": ByNone} {
|
||||
if got := Default(name); got != want {
|
||||
t.Errorf("%s is judged by %q, want %q", name, got, want)
|
||||
}
|
||||
}
|
||||
if ControllerWithin != 60*time.Second || NodeToolsWithin != 60*time.Second || PingWithin != 5*time.Second {
|
||||
t.Fatal("the bounds are to-be 45 §8's: the lease within sixty seconds, a PING answered within five")
|
||||
}
|
||||
}
|
||||
|
||||
// A verdict on the wire names its component as the controller's condition does.
|
||||
func TestAVerdictNamesItsComponent(t *testing.T) {
|
||||
raw, _ := json.Marshal(link.Rollback{Component: Component(ByLease), Outcome: link.RolledBack})
|
||||
if !strings.Contains(string(raw), `"component":"controller"`) || Component(ByPing) != "node-tools" {
|
||||
t.Fatalf("a verdict names its component as %s", raw)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,371 @@
|
||||
package witness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
)
|
||||
|
||||
// What the engine keeps beside a witnessed process (to-be 45 §8): the build before the running one,
|
||||
// and what it knows about the running one — on trial or proved, and what was concluded.
|
||||
//
|
||||
// <root>/<name> the running build, where its unit runs it from
|
||||
// <root>/.witness/<name>/previous/ the build before it, kept until the running one is proved
|
||||
// <root>/.witness/<name>/state.json State
|
||||
//
|
||||
// **Never deleted before the new build is proved**, and deleted once it is: the previous build has
|
||||
// exactly one reader — a restoration — and none once the build after it has been seen healthy.
|
||||
// The running build's directory never moves while it is judged, so its unit is the same unit
|
||||
// throughout, and restoring is two renames and a restart.
|
||||
|
||||
// Dir is where the engine keeps what it knows about a witnessed process.
|
||||
func Dir(root, name string) string { return filepath.Join(root, ".witness", name) }
|
||||
|
||||
func previousDir(root, name string) string { return filepath.Join(Dir(root, name), "previous") }
|
||||
func statePath(root, name string) string { return filepath.Join(Dir(root, name), "state.json") }
|
||||
|
||||
// State is one witnessed process, as the engine keeps it.
|
||||
type State struct {
|
||||
Process string `json:"process"`
|
||||
Witness string `json:"witness"`
|
||||
// Running is the digest of the build at <root>/<name>; Proven, whether it has been seen healthy
|
||||
// (or ran before any witness watched it, or is a build restored — judged once already).
|
||||
Running string `json:"running"`
|
||||
Proven bool `json:"proven"`
|
||||
// Previous is the digest of the build kept to go back to, while Running is on trial.
|
||||
Previous string `json:"previous,omitempty"`
|
||||
// Started is when Running was placed; Watched how long the witness has judged it while it could
|
||||
// ask, Unasked how long it could not. Within is its bound.
|
||||
Started time.Time `json:"started,omitempty"`
|
||||
Watched time.Duration `json:"watched,omitempty"`
|
||||
Unasked time.Duration `json:"unasked,omitempty"`
|
||||
Within time.Duration `json:"within,omitempty"`
|
||||
// NotReversible is why Running may not be rolled back, as its declaration said: the build
|
||||
// before it would run against what this one changed.
|
||||
NotReversible string `json:"not-reversible,omitempty"`
|
||||
// Verdicts are what the witness concluded and still stands: said on every report until the mesh
|
||||
// asks for another build, or a build is proved.
|
||||
Verdicts []link.Rollback `json:"verdicts,omitempty"`
|
||||
// Refused are builds rolled back here: one rollback per build, so a build in this list is not
|
||||
// placed again until a newer one is proved.
|
||||
Refused []string `json:"refused,omitempty"`
|
||||
}
|
||||
|
||||
// OnTrial says whether the running build is still being judged.
|
||||
func (s State) OnTrial() bool {
|
||||
if s.Proven || s.Running == "" {
|
||||
return false
|
||||
}
|
||||
for _, v := range s.Verdicts {
|
||||
if v.From == s.Running {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (s State) refuses(digest string) bool {
|
||||
for _, d := range s.Refused {
|
||||
if d == digest {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Load is what the engine keeps about one process; a zero State when it keeps nothing.
|
||||
func Load(root, name string) (State, error) {
|
||||
raw, err := os.ReadFile(statePath(root, name))
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return State{}, nil
|
||||
}
|
||||
if err != nil {
|
||||
return State{}, err
|
||||
}
|
||||
var s State
|
||||
if err := json.Unmarshal(raw, &s); err != nil {
|
||||
return State{}, fmt.Errorf("what the engine keeps about %s cannot be read: %w", name, err)
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// Save keeps it, whole or not at all.
|
||||
func Save(root string, s State) error {
|
||||
dir := Dir(root, s.Process)
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
return err
|
||||
}
|
||||
body, err := json.MarshalIndent(s, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tmp, err := os.CreateTemp(dir, ".state-*")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer os.Remove(tmp.Name())
|
||||
if _, err := tmp.Write(body); err != nil {
|
||||
tmp.Close()
|
||||
return err
|
||||
}
|
||||
if err := tmp.Sync(); err != nil {
|
||||
tmp.Close()
|
||||
return err
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp.Name(), statePath(root, s.Process))
|
||||
}
|
||||
|
||||
// Forget is a witnessed process no longer declared: everything kept about it goes with it.
|
||||
func Forget(root, name string) error { return os.RemoveAll(Dir(root, name)) }
|
||||
|
||||
// Placing is what the applier is to do with a declared build of a witnessed process.
|
||||
type Placing struct {
|
||||
// Unpack is whether the declared build is to be unpacked at <root>/<name>; false when the build
|
||||
// there already is the one to run.
|
||||
Unpack bool
|
||||
// Detail is what to say about it, when anything.
|
||||
Detail string
|
||||
// Commit records the placement once the build is unpacked and started.
|
||||
Commit func() error
|
||||
}
|
||||
|
||||
// Place readies <root>/<name> for a declared build of a witnessed process, before anything is
|
||||
// unpacked there. `recorded` is the digest the host's record says it placed last, for a process the
|
||||
// engine keeps nothing about yet — every one on the day this ships.
|
||||
//
|
||||
// - the declared build is the one running (restored here, or declared again): nothing is unpacked.
|
||||
// - it was rolled back here and nothing newer has been proved: refused, and the running build stays.
|
||||
// - the running build is proved: it is moved aside as the previous one, and the new one goes on trial.
|
||||
// - the running build is itself on trial: it is discarded, and the previous one stays the one to go
|
||||
// back to — the last build seen healthy, never one that was not.
|
||||
// - nothing runs there yet: a first placement, with nothing to go back to and nothing to judge.
|
||||
func Place(root, name, by, declared, recorded, notReversible string, now time.Time) (Placing, error) {
|
||||
at := filepath.Join(root, name)
|
||||
s, err := Load(root, name)
|
||||
if err != nil {
|
||||
return Placing{}, err
|
||||
}
|
||||
if s.Process == "" {
|
||||
// Nothing kept: what is there is whatever the record says, and it ran before any witness —
|
||||
// it is the build a trial would go back to.
|
||||
s = State{Process: name, Running: recorded, Proven: true}
|
||||
}
|
||||
s.Witness = by
|
||||
present := false
|
||||
if info, err := os.Stat(at); err == nil && info.IsDir() {
|
||||
present = true
|
||||
}
|
||||
|
||||
if present && s.Running == declared {
|
||||
// Asked for the build already running. A restoration followed by the mesh asking for that
|
||||
// same build again ends what was concluded about the build it replaced: the mesh asks for what
|
||||
// runs. What was concluded about this build itself stands.
|
||||
var standing []link.Rollback
|
||||
for _, v := range s.Verdicts {
|
||||
if v.From == s.Running {
|
||||
standing = append(standing, v)
|
||||
}
|
||||
}
|
||||
s.Verdicts = standing
|
||||
return Placing{Commit: func() error { return Save(root, s) }}, nil
|
||||
}
|
||||
if present && s.refuses(declared) {
|
||||
return Placing{
|
||||
Detail: fmt.Sprintf("kept build %s: %s was rolled back on this machine and is not placed again "+
|
||||
"until a newer build has proved itself (novox/hq to-be 45 §8)", short(s.Running), short(declared)),
|
||||
Commit: func() error { return Save(root, s) },
|
||||
}, nil
|
||||
}
|
||||
|
||||
previous := s.Previous
|
||||
switch {
|
||||
case !present || s.Running == "":
|
||||
// A first placement, or a directory that went missing: nothing to keep.
|
||||
if err := os.RemoveAll(at); err != nil {
|
||||
return Placing{}, err
|
||||
}
|
||||
s = State{Process: name, Witness: by, Running: declared, Proven: true, Refused: s.Refused}
|
||||
return Placing{Unpack: true, Commit: func() error { return Save(root, s) }}, nil
|
||||
case s.OnTrial():
|
||||
// The running build has not been seen healthy: it is not what anybody goes back to.
|
||||
if err := os.RemoveAll(at); err != nil {
|
||||
return Placing{}, err
|
||||
}
|
||||
default:
|
||||
if err := os.RemoveAll(previousDir(root, name)); err != nil {
|
||||
return Placing{}, err
|
||||
}
|
||||
if err := os.MkdirAll(Dir(root, name), 0o700); err != nil {
|
||||
return Placing{}, err
|
||||
}
|
||||
if err := os.Rename(at, previousDir(root, name)); err != nil {
|
||||
return Placing{}, fmt.Errorf("cannot keep %s's running build to go back to: %w", name, err)
|
||||
}
|
||||
previous = s.Running
|
||||
}
|
||||
s = State{Process: name, Witness: by, Running: declared, Previous: previous, Started: now.UTC(),
|
||||
Within: Within(by), NotReversible: notReversible, Refused: s.Refused}
|
||||
// Kept as soon as the old build is aside, so a host that stops here knows on its return which
|
||||
// build is where.
|
||||
if err := Save(root, s); err != nil {
|
||||
return Placing{}, err
|
||||
}
|
||||
return Placing{Unpack: true, Commit: func() error { return Save(root, s) }}, nil
|
||||
}
|
||||
|
||||
// Proved is the running build seen healthy: the build before it is deleted — it has no reader now —
|
||||
// and what was concluded and refused before it ends.
|
||||
func Proved(root, name string) error {
|
||||
s, err := Load(root, name)
|
||||
if err != nil || s.Process == "" {
|
||||
return err
|
||||
}
|
||||
if err := os.RemoveAll(previousDir(root, name)); err != nil {
|
||||
return err
|
||||
}
|
||||
s.Proven, s.Previous, s.Verdicts, s.Refused = true, "", nil, nil
|
||||
s.Watched, s.Unasked = 0, 0
|
||||
return Save(root, s)
|
||||
}
|
||||
|
||||
// Runner is how the engine asks the machine's service manager, as the applier does.
|
||||
type Runner func(ctx context.Context, name string, args ...string) (string, error)
|
||||
|
||||
// Restore is the running build not healthy in bound: the previous one put back and started, once —
|
||||
// or, when the running build is declared not reversible or nothing is kept, nothing done and that
|
||||
// said. The verdict is kept, and returned to be said.
|
||||
func Restore(ctx context.Context, root, name, why string, run Runner, now time.Time) (link.Rollback, error) {
|
||||
s, err := Load(root, name)
|
||||
if err != nil {
|
||||
return link.Rollback{}, err
|
||||
}
|
||||
v := link.Rollback{Component: Component(s.Witness), From: s.Running, Why: why, At: now.UTC()}
|
||||
conclude := func(v link.Rollback) (link.Rollback, error) {
|
||||
s.Verdicts = append(s.Verdicts, v)
|
||||
return v, Save(root, s)
|
||||
}
|
||||
switch {
|
||||
case s.NotReversible != "":
|
||||
v.Outcome = link.NotReversible
|
||||
v.Why = why + "; not rolled back: this build is declared not reversible (" + s.NotReversible +
|
||||
"), so the build before it would run against what it changed. It is left running. A person decides"
|
||||
return conclude(v)
|
||||
case s.Previous == "":
|
||||
v.Outcome = link.NothingToRestore
|
||||
v.Why = why + "; no build before it is kept on this machine"
|
||||
return conclude(v)
|
||||
}
|
||||
if _, err := os.Stat(previousDir(root, name)); err != nil {
|
||||
v.Outcome = link.NothingToRestore
|
||||
v.Why = fmt.Sprintf("%s; the build before it (%s) is not where it was kept: %v", why, short(s.Previous), err)
|
||||
return conclude(v)
|
||||
}
|
||||
|
||||
unit := name + ".service"
|
||||
// Stopped first, so the failing build is not running while its files are moved; a stop that fails
|
||||
// is not fatal — the restart below replaces whatever runs.
|
||||
_, _ = run(ctx, "systemctl", "stop", unit)
|
||||
at := filepath.Join(root, name)
|
||||
failed := filepath.Join(Dir(root, name), "failed")
|
||||
if err := os.RemoveAll(failed); err != nil {
|
||||
return restoreFailed(root, s, v, err)
|
||||
}
|
||||
if err := os.Rename(at, failed); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
return restoreFailed(root, s, v, err)
|
||||
}
|
||||
if err := os.Rename(previousDir(root, name), at); err != nil {
|
||||
// Put back what was there, so the machine is no worse than before the attempt.
|
||||
_ = os.Rename(failed, at)
|
||||
return restoreFailed(root, s, v, err)
|
||||
}
|
||||
_ = os.RemoveAll(failed)
|
||||
|
||||
v.To, v.Outcome = s.Previous, link.RolledBack
|
||||
s.Refused = append(s.Refused, s.Running)
|
||||
// The restored build was proved before; it is not judged a second time. One rollback per build.
|
||||
s.Running, s.Previous, s.Proven = s.Previous, "", true
|
||||
if _, err := run(ctx, "systemctl", "restart", unit); err != nil {
|
||||
v.Outcome = link.RestoreFailed
|
||||
v.Why = fmt.Sprintf("%s; the build before it (%s) was put back and would not start: %v", why, short(v.To), err)
|
||||
}
|
||||
return conclude(v)
|
||||
}
|
||||
|
||||
func restoreFailed(root string, s State, v link.Rollback, err error) (link.Rollback, error) {
|
||||
v.Outcome = link.RestoreFailed
|
||||
v.Why = fmt.Sprintf("%s; putting the build before it (%s) back failed: %v", v.Why, short(s.Previous), err)
|
||||
s.Verdicts = append(s.Verdicts, v)
|
||||
return v, Save(root, s)
|
||||
}
|
||||
|
||||
// Conclude keeps a verdict that restores nothing — a build that could not be judged at all.
|
||||
func Conclude(root, name string, v link.Rollback) error {
|
||||
s, err := Load(root, name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.Verdicts = append(s.Verdicts, v)
|
||||
return Save(root, s)
|
||||
}
|
||||
|
||||
// Standing is every verdict that still stands, on every witnessed process under root, in a stable
|
||||
// order — what every report says.
|
||||
func Standing(root string) []link.Rollback {
|
||||
var out []link.Rollback
|
||||
for _, s := range all(root) {
|
||||
out = append(out, s.Verdicts...)
|
||||
}
|
||||
sort.SliceStable(out, func(i, j int) bool {
|
||||
if out[i].Component != out[j].Component {
|
||||
return out[i].Component < out[j].Component
|
||||
}
|
||||
return out[i].At.Before(out[j].At)
|
||||
})
|
||||
return out
|
||||
}
|
||||
|
||||
// Trials is every witnessed process whose running build is being judged.
|
||||
func Trials(root string) []State {
|
||||
var out []State
|
||||
for _, s := range all(root) {
|
||||
if s.OnTrial() {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func all(root string) []State {
|
||||
entries, err := os.ReadDir(filepath.Join(root, ".witness"))
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
var out []State
|
||||
for _, e := range entries {
|
||||
if !e.IsDir() {
|
||||
continue
|
||||
}
|
||||
if s, err := Load(root, e.Name()); err == nil && s.Process != "" {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Process < out[j].Process })
|
||||
return out
|
||||
}
|
||||
|
||||
func short(digest string) string {
|
||||
if len(digest) > len("sha256:")+12 {
|
||||
return digest[:len("sha256:")+12]
|
||||
}
|
||||
return digest
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
package witness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
)
|
||||
|
||||
// Watcher judges every witnessed build on trial on this machine, every Every, until each is proved or
|
||||
// concluded (to-be 45 §8).
|
||||
//
|
||||
// **Time counts only while it could ask.** A build's bound is spent only on looks that got an answer
|
||||
// from the bus — the lease read, or the PING delivered — so a machine whose own link is down, or a
|
||||
// bus that refuses the question, never rolls a build back for the host's own trouble. A look that
|
||||
// could not ask counts towards GiveUp instead, and at GiveUp the build is said to be unwitnessed:
|
||||
// neither proved nor rolled back, and said.
|
||||
type Watcher struct {
|
||||
Root string
|
||||
// Node is this machine's node name, the instance its runtime answers PING as; Host its hostname,
|
||||
// as the controller's lease names its machine.
|
||||
Node, Host string
|
||||
// Asker is the link open now, or nil.
|
||||
Asker func() link.Asker
|
||||
Run Runner
|
||||
// Hold runs a change to what is placed on the machine in turn with every apply: the restoration
|
||||
// moves the directory an apply writes into.
|
||||
Hold func(func())
|
||||
// Concluded is told every verdict, once, as it is reached: to say it now rather than at the next
|
||||
// report.
|
||||
Concluded func(link.Rollback)
|
||||
Say func(string)
|
||||
Now func() time.Time
|
||||
Every time.Duration
|
||||
}
|
||||
|
||||
// Watch looks every Every until ctx ends.
|
||||
func (w *Watcher) Watch(ctx context.Context) {
|
||||
every := w.Every
|
||||
if every <= 0 {
|
||||
every = Every
|
||||
}
|
||||
ticker := time.NewTicker(every)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
w.Look(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Look judges every build on trial once.
|
||||
func (w *Watcher) Look(ctx context.Context) {
|
||||
for _, s := range Trials(w.Root) {
|
||||
w.look(ctx, s)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Watcher) look(ctx context.Context, s State) {
|
||||
every := w.Every
|
||||
if every <= 0 {
|
||||
every = Every
|
||||
}
|
||||
now := w.now()
|
||||
healthy, why, err := w.judge(ctx, s, now)
|
||||
|
||||
w.hold(func() {
|
||||
// Read again in turn: an apply may have placed another build meanwhile, which starts its own
|
||||
// trial — this look was about the one before it.
|
||||
current, loadErr := Load(w.Root, s.Process)
|
||||
if loadErr != nil || current.Running != s.Running || !current.OnTrial() {
|
||||
return
|
||||
}
|
||||
switch {
|
||||
case err != nil:
|
||||
current.Unasked += every
|
||||
if current.Unasked < GiveUp {
|
||||
_ = Save(w.Root, current)
|
||||
return
|
||||
}
|
||||
v := link.Rollback{Component: Component(current.Witness), From: current.Running, Outcome: link.Unwitnessed,
|
||||
Why: fmt.Sprintf("its health could not be judged for %s: %v. The build before it is kept", GiveUp, err),
|
||||
At: now.UTC()}
|
||||
current.Verdicts = append(current.Verdicts, v)
|
||||
if Save(w.Root, current) == nil {
|
||||
w.concluded(v)
|
||||
}
|
||||
case healthy:
|
||||
if Proved(w.Root, s.Process) == nil {
|
||||
w.say(fmt.Sprintf("%s %s is healthy: %s; the build before it is deleted", s.Process, short(s.Running), why))
|
||||
}
|
||||
default:
|
||||
current.Watched += every
|
||||
if current.Watched < current.Within {
|
||||
_ = Save(w.Root, current)
|
||||
return
|
||||
}
|
||||
if err := Save(w.Root, current); err != nil {
|
||||
return
|
||||
}
|
||||
bound := fmt.Sprintf("%s %s was not healthy within %s of starting: %s",
|
||||
s.Process, short(s.Running), current.Within, why)
|
||||
v, err := Restore(ctx, w.Root, s.Process, bound, w.Run, now)
|
||||
if err != nil {
|
||||
w.say(fmt.Sprintf("%s; and what was concluded could not be kept: %v", bound, err))
|
||||
}
|
||||
w.concluded(v)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// judge asks once. An error is a look that could not ask; otherwise healthy, and why not when not.
|
||||
func (w *Watcher) judge(ctx context.Context, s State, now time.Time) (bool, string, error) {
|
||||
var asker link.Asker
|
||||
if w.Asker != nil {
|
||||
asker = w.Asker()
|
||||
}
|
||||
if asker == nil {
|
||||
return false, "", errors.New("this host is not linked to the bus")
|
||||
}
|
||||
asking, cancel := context.WithTimeout(ctx, PingWithin)
|
||||
defer cancel()
|
||||
switch s.Witness {
|
||||
case ByLease:
|
||||
value, found, err := asker.ReadKey(asking, LeaseBucket, LeaseKey)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
if !found {
|
||||
return false, "nobody holds the controller's lease", nil
|
||||
}
|
||||
lease, err := ParseLease(value)
|
||||
if err != nil {
|
||||
return false, err.Error(), nil
|
||||
}
|
||||
held, why := lease.HeldBySince(w.Host, s.Started, now)
|
||||
return held, why, nil
|
||||
case ByPing:
|
||||
err := asker.Ping(asking, s.Process, w.Node)
|
||||
switch {
|
||||
case err == nil:
|
||||
return true, "it answered PING", nil
|
||||
case errors.Is(err, link.ErrNoAnswer):
|
||||
return false, err.Error(), nil
|
||||
default:
|
||||
return false, "", err
|
||||
}
|
||||
}
|
||||
return false, "", fmt.Errorf("%s names no witness this host knows (%q)", s.Process, s.Witness)
|
||||
}
|
||||
|
||||
func (w *Watcher) hold(f func()) {
|
||||
if w.Hold == nil {
|
||||
f()
|
||||
return
|
||||
}
|
||||
w.Hold(f)
|
||||
}
|
||||
|
||||
func (w *Watcher) now() time.Time {
|
||||
if w.Now == nil {
|
||||
return time.Now()
|
||||
}
|
||||
return w.Now()
|
||||
}
|
||||
|
||||
func (w *Watcher) say(line string) {
|
||||
if w.Say != nil {
|
||||
w.Say(line)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Watcher) concluded(v link.Rollback) {
|
||||
w.say(fmt.Sprintf("%s %s: %s — %s", v.Component, v.Outcome, short(v.From), v.Why))
|
||||
if w.Concluded != nil {
|
||||
w.Concluded(v)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,346 @@
|
||||
package witness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
)
|
||||
|
||||
// Each rollback path, induced against a real directory: a controller that never takes the lease, a
|
||||
// runtime that never answers PING, a build declared not reversible; and a healthy update, which
|
||||
// deletes nothing before it is proved and the build before it once it is (novox/hq to-be 45 §8).
|
||||
|
||||
const (
|
||||
buildA = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
buildB = "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
||||
buildC = "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
|
||||
)
|
||||
|
||||
// place does what the applier does with a declared build: ask Place, unpack when told, commit.
|
||||
func place(t *testing.T, root, name, by, build, recorded, notReversible string, at time.Time) Placing {
|
||||
t.Helper()
|
||||
p, err := Place(root, name, by, build, recorded, notReversible, at)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if p.Unpack {
|
||||
dir := filepath.Join(root, name)
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(dir, "build"), []byte(build), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := p.Commit(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
func running(t *testing.T, root, name string) string {
|
||||
t.Helper()
|
||||
raw, err := os.ReadFile(filepath.Join(root, name, "build"))
|
||||
if err != nil {
|
||||
return "nothing"
|
||||
}
|
||||
return string(raw)
|
||||
}
|
||||
|
||||
func kept(root, name string) bool {
|
||||
_, err := os.Stat(previousDir(root, name))
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// asker answers as the build it is told is running would.
|
||||
type asker struct {
|
||||
mu sync.Mutex
|
||||
lease func() ([]byte, bool, error)
|
||||
ping func() error
|
||||
asked int
|
||||
}
|
||||
|
||||
func (a *asker) ReadKey(context.Context, string, string) ([]byte, bool, error) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.asked++
|
||||
return a.lease()
|
||||
}
|
||||
|
||||
func (a *asker) Ping(context.Context, string, string) error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.asked++
|
||||
return a.ping()
|
||||
}
|
||||
|
||||
type machine struct {
|
||||
commands []string
|
||||
failRestart bool
|
||||
}
|
||||
|
||||
func (m *machine) run(_ context.Context, name string, args ...string) (string, error) {
|
||||
m.commands = append(m.commands, name+" "+strings.Join(args, " "))
|
||||
if m.failRestart && len(args) > 0 && args[0] == "restart" {
|
||||
return "", fmt.Errorf("exit status 1")
|
||||
}
|
||||
return "", nil
|
||||
}
|
||||
|
||||
func watcher(root string, a link.Asker, m *machine, clock *time.Time, verdicts *[]link.Rollback) *Watcher {
|
||||
return &Watcher{Root: root, Node: "anchor", Host: "anchor",
|
||||
Asker: func() link.Asker {
|
||||
if a == nil {
|
||||
return nil
|
||||
}
|
||||
return a
|
||||
},
|
||||
Run: m.run, Now: func() time.Time { return *clock },
|
||||
Concluded: func(v link.Rollback) { *verdicts = append(*verdicts, v) },
|
||||
Every: Every,
|
||||
}
|
||||
}
|
||||
|
||||
// look runs the watcher n times, the clock moving Every each time.
|
||||
func look(w *Watcher, clock *time.Time, n int) {
|
||||
for i := 0; i < n; i++ {
|
||||
*clock = clock.Add(Every)
|
||||
w.Look(context.Background())
|
||||
}
|
||||
}
|
||||
|
||||
func leaseHeldBy(instance string, taken time.Time, clock *time.Time) func() ([]byte, bool, error) {
|
||||
return func() ([]byte, bool, error) {
|
||||
return []byte(fmt.Sprintf(`{"instance":%q,"host":"anchor","epoch":9,"taken":%q,"renewed":%q}`,
|
||||
instance, taken.Format(time.RFC3339Nano), clock.Add(-time.Second).Format(time.RFC3339Nano))), true, nil
|
||||
}
|
||||
}
|
||||
|
||||
// A controller that starts and never takes the lease: restored to the build before, once, said once.
|
||||
func TestAControllerThatNeverTakesTheLeaseIsRolledBackOnce(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
clock := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
|
||||
place(t, root, "mesh-controller", ByLease, buildA, "", "", clock)
|
||||
place(t, root, "mesh-controller", ByLease, buildB, buildA, "", clock)
|
||||
if running(t, root, "mesh-controller") != buildB || !kept(root, "mesh-controller") {
|
||||
t.Fatal("the new controller is not running with the build before it kept beside it")
|
||||
}
|
||||
|
||||
// The old instance's lease, taken an hour ago, is all the bus ever shows.
|
||||
a := &asker{lease: leaseHeldBy("controller@anchor pid 1 since earlier", clock.Add(-time.Hour), &clock)}
|
||||
m := &machine{}
|
||||
var verdicts []link.Rollback
|
||||
w := watcher(root, a, m, &clock, &verdicts)
|
||||
|
||||
look(w, &clock, int(ControllerWithin/Every)-1)
|
||||
if len(verdicts) != 0 || running(t, root, "mesh-controller") != buildB {
|
||||
t.Fatalf("rolled back before its bound: %+v", verdicts)
|
||||
}
|
||||
look(w, &clock, 1)
|
||||
if len(verdicts) != 1 || verdicts[0].Outcome != link.RolledBack || verdicts[0].From != buildB || verdicts[0].To != buildA ||
|
||||
verdicts[0].Component != link.ComponentController {
|
||||
t.Fatalf("the verdict is %+v, want controller rolled back from B to A", verdicts)
|
||||
}
|
||||
if running(t, root, "mesh-controller") != buildA {
|
||||
t.Fatalf("the controller running is %s, want the build before", running(t, root, "mesh-controller"))
|
||||
}
|
||||
if strings.Join(m.commands, "; ") != "systemctl stop mesh-controller.service; systemctl restart mesh-controller.service" {
|
||||
t.Fatalf("the machine was asked %v", m.commands)
|
||||
}
|
||||
if !strings.Contains(verdicts[0].Why, "not healthy within 1m0s") || !strings.Contains(verdicts[0].Why, "before the new build started") {
|
||||
t.Fatalf("the verdict does not say why: %s", verdicts[0].Why)
|
||||
}
|
||||
|
||||
// Once: the restored build is not judged again, and nothing more is said or done.
|
||||
look(w, &clock, 30)
|
||||
if len(verdicts) != 1 || len(m.commands) != 2 {
|
||||
t.Fatalf("judged again after a rollback: %d verdicts, %v", len(verdicts), m.commands)
|
||||
}
|
||||
// Standing, so every report says it — until the mesh asks for another build.
|
||||
if s := Standing(root); len(s) != 1 || s[0].From != buildB {
|
||||
t.Fatalf("what every report says is %+v", s)
|
||||
}
|
||||
// The mesh still declares B: refused, A kept running, the verdict still standing.
|
||||
p := place(t, root, "mesh-controller", ByLease, buildB, buildB, "", clock)
|
||||
if p.Unpack || running(t, root, "mesh-controller") != buildA || !strings.Contains(p.Detail, "rolled back") {
|
||||
t.Fatalf("a rolled-back build was placed again: %+v, running %s", p, running(t, root, "mesh-controller"))
|
||||
}
|
||||
if len(Standing(root)) != 1 || len(Trials(root)) != 0 {
|
||||
t.Fatal("refusing the rolled-back build changed what stands or started a trial")
|
||||
}
|
||||
// A newer build is a new trial, and ends what was concluded once it is proved.
|
||||
place(t, root, "mesh-controller", ByLease, buildC, buildB, "", clock)
|
||||
if running(t, root, "mesh-controller") != buildC || len(Trials(root)) != 1 {
|
||||
t.Fatal("a newer build after a rollback was not placed on trial")
|
||||
}
|
||||
if err := Proved(root, "mesh-controller"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(Standing(root)) != 0 || kept(root, "mesh-controller") {
|
||||
t.Fatal("a newer build proved and the rollback still stands, or the build before it is still kept")
|
||||
}
|
||||
}
|
||||
|
||||
// A runtime that never answers PING: restored, once.
|
||||
func TestARuntimeThatNeverAnswersIsRolledBackOnce(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
clock := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
|
||||
place(t, root, "node-tools", ByPing, buildA, "", "", clock)
|
||||
place(t, root, "node-tools", ByPing, buildB, buildA, "", clock)
|
||||
|
||||
a := &asker{ping: func() error { return fmt.Errorf("%w: no node-tools on this machine is on the bus", link.ErrNoAnswer) }}
|
||||
m := &machine{}
|
||||
var verdicts []link.Rollback
|
||||
w := watcher(root, a, m, &clock, &verdicts)
|
||||
look(w, &clock, int(NodeToolsWithin/Every)+20)
|
||||
|
||||
if len(verdicts) != 1 || verdicts[0].Outcome != link.RolledBack || verdicts[0].Component != link.ComponentNodeTools {
|
||||
t.Fatalf("the verdict is %+v, want node-tools rolled back once", verdicts)
|
||||
}
|
||||
if running(t, root, "node-tools") != buildA || kept(root, "node-tools") {
|
||||
t.Fatal("the runtime running is not the build before, or a copy of it is still kept")
|
||||
}
|
||||
}
|
||||
|
||||
// A healthy update: nothing deleted before it is proved; the build before it deleted once it is.
|
||||
func TestAHealthyUpdateDeletesNothingUntilItIsProved(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
clock := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
|
||||
// The controller placed by a host from before any witness: a record, and no state.
|
||||
if err := os.MkdirAll(filepath.Join(root, "mesh-controller"), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(root, "mesh-controller", "build"), []byte(buildA), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
place(t, root, "mesh-controller", ByLease, buildB, buildA, "", clock)
|
||||
started := clock
|
||||
|
||||
a := &asker{lease: func() ([]byte, bool, error) { return nil, false, nil }}
|
||||
m := &machine{}
|
||||
var verdicts []link.Rollback
|
||||
w := watcher(root, a, m, &clock, &verdicts)
|
||||
look(w, &clock, 5)
|
||||
if !kept(root, "mesh-controller") {
|
||||
t.Fatal("the build before was deleted while the new one was still on trial")
|
||||
}
|
||||
// A third build while the second is on trial: the one kept is still the one seen healthy.
|
||||
place(t, root, "mesh-controller", ByLease, buildC, buildB, "", clock)
|
||||
if s, _ := Load(root, "mesh-controller"); s.Previous != buildA {
|
||||
t.Fatalf("the build kept to go back to is %s, want A — the last one seen healthy", s.Previous)
|
||||
}
|
||||
started = clock
|
||||
a.lease = leaseHeldBy("controller@anchor pid 3 since now", started.Add(4*time.Second), &clock)
|
||||
look(w, &clock, 2)
|
||||
if len(verdicts) != 0 || len(m.commands) != 0 {
|
||||
t.Fatalf("a healthy update was judged or acted on: %+v %v", verdicts, m.commands)
|
||||
}
|
||||
if kept(root, "mesh-controller") || len(Trials(root)) != 0 || running(t, root, "mesh-controller") != buildC {
|
||||
t.Fatal("the build before was not deleted once the new one was proved")
|
||||
}
|
||||
}
|
||||
|
||||
// A build declared not reversible is never rolled back: the build before never starts against what it
|
||||
// changed, and the verdict is urgent and stands.
|
||||
func TestANotReversibleBuildIsNeverRolledBack(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
clock := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
|
||||
place(t, root, "mesh-controller", ByLease, buildA, "", "", clock)
|
||||
place(t, root, "mesh-controller", ByLease, buildB, buildA, "migration 0073 cannot be undone", clock)
|
||||
|
||||
a := &asker{lease: func() ([]byte, bool, error) { return nil, false, nil }}
|
||||
m := &machine{}
|
||||
var verdicts []link.Rollback
|
||||
w := watcher(root, a, m, &clock, &verdicts)
|
||||
look(w, &clock, int(ControllerWithin/Every)+20)
|
||||
|
||||
if len(verdicts) != 1 || verdicts[0].Outcome != link.NotReversible || verdicts[0].To != "" {
|
||||
t.Fatalf("the verdict is %+v, want not-reversible with nothing restored", verdicts)
|
||||
}
|
||||
if !strings.Contains(verdicts[0].Why, "migration 0073 cannot be undone") {
|
||||
t.Fatalf("the verdict does not say why it may not be rolled back: %s", verdicts[0].Why)
|
||||
}
|
||||
if len(m.commands) != 0 || running(t, root, "mesh-controller") != buildB {
|
||||
t.Fatalf("the build before was started against the newer data: %v, running %s", m.commands,
|
||||
running(t, root, "mesh-controller"))
|
||||
}
|
||||
if len(Standing(root)) != 1 {
|
||||
t.Fatal("the not-reversible verdict does not stand")
|
||||
}
|
||||
}
|
||||
|
||||
// A witness that cannot ask counts nothing against the build, and says so at GiveUp.
|
||||
func TestAWitnessThatCannotAskNeverRollsBack(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
clock := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
|
||||
place(t, root, "mesh-controller", ByLease, buildA, "", "", clock)
|
||||
place(t, root, "mesh-controller", ByLease, buildB, buildA, "", clock)
|
||||
|
||||
a := &asker{lease: func() ([]byte, bool, error) {
|
||||
return nil, false, fmt.Errorf("%w: this host's grant does not name the lease", link.ErrCannotAsk)
|
||||
}}
|
||||
m := &machine{}
|
||||
var verdicts []link.Rollback
|
||||
w := watcher(root, a, m, &clock, &verdicts)
|
||||
look(w, &clock, int(GiveUp/Every)-1)
|
||||
if len(verdicts) != 0 || len(m.commands) != 0 || running(t, root, "mesh-controller") != buildB {
|
||||
t.Fatalf("a build was judged on questions that were never asked: %+v %v", verdicts, m.commands)
|
||||
}
|
||||
look(w, &clock, 1)
|
||||
if len(verdicts) != 1 || verdicts[0].Outcome != link.Unwitnessed || len(m.commands) != 0 {
|
||||
t.Fatalf("the verdict at GiveUp is %+v, want unwitnessed with nothing done", verdicts)
|
||||
}
|
||||
if !kept(root, "mesh-controller") {
|
||||
t.Fatal("the build before was deleted though the new one was never seen healthy")
|
||||
}
|
||||
|
||||
// And a host with no link at all is the same: nothing counted.
|
||||
root2 := t.TempDir()
|
||||
place(t, root2, "node-tools", ByPing, buildA, "", "", clock)
|
||||
place(t, root2, "node-tools", ByPing, buildB, buildA, "", clock)
|
||||
var none []link.Rollback
|
||||
w2 := watcher(root2, nil, m, &clock, &none)
|
||||
look(w2, &clock, int(NodeToolsWithin/Every)*3)
|
||||
if len(none) != 0 || running(t, root2, "node-tools") != buildB {
|
||||
t.Fatal("a build was rolled back while this host had no link to ask with")
|
||||
}
|
||||
}
|
||||
|
||||
// A restoration whose build will not start is said as such — not as a rollback that worked.
|
||||
func TestARestorationThatDoesNotStartIsSaid(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
clock := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
|
||||
place(t, root, "node-tools", ByPing, buildA, "", "", clock)
|
||||
place(t, root, "node-tools", ByPing, buildB, buildA, "", clock)
|
||||
v, err := Restore(context.Background(), root, "node-tools", "it never answered", (&machine{failRestart: true}).run, clock)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if v.Outcome != link.RestoreFailed || running(t, root, "node-tools") != buildA {
|
||||
t.Fatalf("the verdict is %+v, running %s", v, running(t, root, "node-tools"))
|
||||
}
|
||||
}
|
||||
|
||||
// What the engine keeps survives the engine: a host that stands aside mid-trial resumes it.
|
||||
func TestATrialSurvivesTheHostRestarting(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
clock := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
|
||||
place(t, root, "node-tools", ByPing, buildA, "", "", clock)
|
||||
place(t, root, "node-tools", ByPing, buildB, buildA, "", clock)
|
||||
a := &asker{ping: func() error { return link.ErrNoAnswer }}
|
||||
var verdicts []link.Rollback
|
||||
half := int(NodeToolsWithin/Every) / 2
|
||||
look(watcher(root, a, &machine{}, &clock, &verdicts), &clock, half)
|
||||
// A new watcher, as a successor host would start.
|
||||
look(watcher(root, a, &machine{}, &clock, &verdicts), &clock, int(NodeToolsWithin/Every)-half)
|
||||
if len(verdicts) != 1 {
|
||||
t.Fatalf("the bound was not carried across the host restarting: %+v", verdicts)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user