diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 89cd8f0..fe74d88 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -457,10 +457,10 @@ func short(digest string) string { // them again — and everything the mesh declared read as no longer declared and removed. Even the // very declaration the mesh last sent, applied from a file, would plan to remove the foundation. // `apply FILE` is for a machine the mesh has not spoken to, and is refused saying so. -func refuseStale(known store.State, kept store.Declared, keptErr error, digest string, from provenance) error { +func refuseStale(known store.State, kept store.Declared, keptErr error, digest string, from provenance, sequence int64) error { switch from { case fromDeclared: - return nil + return refuseOlder(kept, keptErr, sequence) case fromBundle: if known.Genesis == nil || known.Genesis.Digest == digest { return nil @@ -557,7 +557,7 @@ func runApply(ctx context.Context, opts options, d *declaration.Declaration, raw if err := apply.CheckMode(known, d); err != nil { return err } - if err := refuseStale(known, kept, keptErr, digest, from); err != nil { + if err := refuseStale(known, kept, keptErr, digest, from, d.Sequence); err != nil { return err } @@ -1594,3 +1594,32 @@ func adoptDeliveredMembership(identityPath string, mine *identity.Identity, say say(fmt.Sprintf("moving to the %s bus at %s — restarting to dial it", next.Transport, next.Broker)) os.Exit(0) } + +// refuseOlder refuses a declaration from the mesh that is older than the one this node holds. +// +// **By sequence, not by arrival** (novox/hq 04-ISSUES/107). What the host kept is the last thing +// the mesh said, signed; a declaration whose sequence is lower was composed before it, whatever +// order they arrived in — a backlog drained after the node was away, or a broker that split a burst. +// Applying it would make the machine into something the mesh had already moved past, which is the +// incident of issue 104 by another door. +// +// Only when both sides claim an order. A declaration with no sequence is one an older controller +// sent, and one kept with no sequence is one this host received before it understood them; in either +// case there is no order to compare, and refusing on a guess would strand the node the moment the +// controller is older than the host. Equal is the same declaration again, which reconciling is for. +func refuseOlder(kept store.Declared, keptErr error, sequence int64) error { + if sequence == 0 || keptErr != nil { + return nil + } + last, err := declaration.ParseTrusted(kept.Declaration) + if err != nil || last.Sequence == 0 { + return nil + } + if sequence < last.Sequence { + return fmt.Errorf("this declaration is older than what the mesh last said to this node: it "+ + "is sequence %d, and the one kept here is %d. It arrived late — a backlog, or a broker "+ + "that split a burst — and applying it would make this machine into something the mesh has "+ + "already moved past. Refused whole; nothing was applied", sequence, last.Sequence) + } + return nil +} diff --git a/cmd/mesh-host/order_test.go b/cmd/mesh-host/order_test.go new file mode 100644 index 0000000..0380aa4 --- /dev/null +++ b/cmd/mesh-host/order_test.go @@ -0,0 +1,58 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/novox/mesh-host/internal/store" +) + +// A declaration carries no order, so a host cannot tell an older one from a newer (novox/hq +// 04-ISSUES/107). Its only identity was the digest of its bytes: "not the last" could be said, +// "older" could not. + +func keptWith(t *testing.T, sequence int64) store.Declared { + t.Helper() + body, err := json.Marshal(map[string]any{ + "declaration": 1, "resources": []any{}, "owns_nothing": true, "sequence": sequence, + }) + if err != nil { + t.Fatal(err) + } + return store.Declared{Declaration: body, Signature: []byte("x")} +} + +func TestAnOlderDeclarationFromTheMeshIsRefused(t *testing.T) { + err := refuseOlder(keptWith(t, 7), nil, 5) + if err == nil { + t.Fatal("sequence 5 was accepted over a kept 7") + } + if !strings.Contains(err.Error(), "older") || !strings.Contains(err.Error(), "nothing was applied") { + t.Fatalf("the refusal does not say what it is: %v", err) + } +} + +func TestANewerOrEqualDeclarationIsNot(t *testing.T) { + if err := refuseOlder(keptWith(t, 7), nil, 8); err != nil { + t.Fatalf("sequence 8 was refused over a kept 7: %v", err) + } + // Equal is the same declaration again, which reconciling is for. + if err := refuseOlder(keptWith(t, 7), nil, 7); err != nil { + t.Fatalf("the same sequence was refused: %v", err) + } +} + +func TestNoOrderClaimedMeansNoOrderCompared(t *testing.T) { + // An older controller sends none; a host that received before it understood them kept none. + // Refusing on a guess would strand a node the moment the controller is older than the host. + if err := refuseOlder(keptWith(t, 7), nil, 0); err != nil { + t.Fatalf("a declaration claiming no order was refused: %v", err) + } + if err := refuseOlder(keptWith(t, 0), nil, 3); err != nil { + t.Fatalf("a declaration was refused against a kept one that claimed no order: %v", err) + } + if err := refuseOlder(store.Declared{}, store.ErrNothingDeclared, 3); err != nil { + t.Fatalf("a first declaration was refused: %v", err) + } +} diff --git a/internal/declaration/declaration.go b/internal/declaration/declaration.go index d0c6e63..d665e7d 100644 --- a/internal/declaration/declaration.go +++ b/internal/declaration/declaration.go @@ -1137,6 +1137,19 @@ type Declaration struct { // converged node — which is every node the mesh raised before adoption existed, and so the // only form an older controller ever sends (novox/hq ADR 0100). Adoption *Adoption + + // Sequence orders this declaration against every other the mesh has sent this node: each + // send is one higher than the last, assigned under the control plane's hold on the node + // (novox/hq 04-ISSUES/107). Zero is a declaration that carries no order — every one an older + // controller sent, and the bundle genesis applies — and a host makes no ordering claim about + // one of those. + // + // **The one property a declaration needs that its signature does not give it.** A signature + // says the mesh sent this; it cannot say the mesh sent it AFTER the one the host is holding. + // Before this, "older" was inferred from arrival within a batch and a 750ms window, and a + // backlog longer than the batch, or a slow broker, applied a declaration the mesh had already + // superseded. + Sequence int64 } // Adoption is a node's mode, as the controller records it: the node is adopted, and these are @@ -1274,6 +1287,9 @@ type envelope struct { // a bug quietly strip a machine. OwnsNothing bool `json:"owns_nothing,omitempty"` Resources []json.RawMessage `json:"resources"` + // Sequence is optional on the wire, so a controller that does not send one is still + // understood: absent reads as zero, which is "no ordering claimed" rather than "first". + Sequence int64 `json:"sequence,omitempty"` } func parse(raw []byte, allowActions bool) (*Declaration, error) { @@ -1290,7 +1306,7 @@ func parse(raw []byte, allowActions bool) (*Declaration, error) { env.Version, Version)}} } - d := &Declaration{Version: env.Version, For: env.For, Adoption: env.Adoption} + d := &Declaration{Version: env.Version, For: env.For, Adoption: env.Adoption, Sequence: env.Sequence} var problems []string if len(env.Resources) == 0 && !env.OwnsNothing { diff --git a/internal/link/order_test.go b/internal/link/order_test.go new file mode 100644 index 0000000..64df68a --- /dev/null +++ b/internal/link/order_test.go @@ -0,0 +1,50 @@ +package link + +import ( + "encoding/json" + "testing" + "time" +) + +// The drain picked the last to arrive. A backlog longer than the batch, or a broker that split a +// burst, delivered a superseded declaration last (novox/hq 04-ISSUES/107). + +func sequenced(t *testing.T, n int64) *said { + t.Helper() + inner, err := json.Marshal(map[string]any{"declaration": 1, "resources": []any{}, "sequence": n}) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(Signed{Declaration: inner, Signature: []byte("s")}) + if err != nil { + t.Fatal(err) + } + return &said{body: body} +} + +func TestTheDrainKeepsTheHighestSequenceNotTheLastToArrive(t *testing.T) { + waiting := make(chan Declaration, 8) + waiting <- sequenced(t, 9) + waiting <- sequenced(t, 4) // arrived last, composed earlier + latest, aside := newest(waiting, sequenced(t, 8), 30*time.Millisecond) + if got := sequenceOf(latest.Body()); got != 9 { + t.Fatalf("the drain kept sequence %d, and 9 was waiting", got) + } + if len(aside) != 2 { + t.Fatalf("%d set aside, wanted 2 (the 8 and the late 4)", len(aside)) + } +} + +func TestWithoutSequencesTheLastToArriveStillWins(t *testing.T) { + // The behaviour this had before, kept for a controller that sends no order. + apply, aside := newest(arriving("two", "three"), &said{body: []byte("one")}, 30*time.Millisecond) + if string(apply.Body()) != "three" || len(aside) != 2 { + t.Fatalf("applied %q with %d set aside", apply.Body(), len(aside)) + } +} + +func TestAnUnreadableBodyClaimsNoOrder(t *testing.T) { + if got := sequenceOf([]byte("not json")); got != 0 { + t.Fatalf("garbage claimed sequence %d", got) + } +} diff --git a/internal/link/run.go b/internal/link/run.go index fb28c00..ae1f28f 100644 --- a/internal/link/run.go +++ b/internal/link/run.go @@ -300,6 +300,16 @@ func newest(arriving <-chan Declaration, first Declaration, window time.Duration if !ok { return latest, superseded } + // **By sequence when both carry one, by arrival when either does not** (novox/hq + // 04-ISSUES/107). Arrival is what this window had to go on, and it is wrong exactly + // when it matters — a backlog drained out of order. A declaration that says where it + // stands is believed over when it turned up; one that does not is the older + // controller's, and arrival is all there is. + if sequenceOf(next.Body()) < sequenceOf(latest.Body()) && + sequenceOf(next.Body()) > 0 && sequenceOf(latest.Body()) > 0 { + superseded = append(superseded, next) + continue + } superseded = append(superseded, latest) latest = next case <-time.After(window): @@ -308,6 +318,24 @@ func newest(arriving <-chan Declaration, first Declaration, window time.Duration } } +// sequenceOf is the order a signed declaration claims, or zero when it claims none or cannot be +// read. Read from the envelope alone; the signature is verified later, when the winner is applied, +// and a forged message that lied about its sequence would only set aside real ones — which are +// reported as set aside, and the next push sends the current one again. +func sequenceOf(body []byte) int64 { + var signed Signed + if err := json.Unmarshal(body, &signed); err != nil { + return 0 + } + var d struct { + Sequence int64 `json:"sequence"` + } + if err := json.Unmarshal(signed.Declaration, &d); err != nil { + return 0 + } + return d.Sequence +} + // declaredIn is the id a signed declaration carries, for a report about one that was not applied. // Empty if the message is not one — a forged or garbled message is refused by handleBody when its // turn comes; here it is only named.