diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/end.go b/modules/mesh-enrolment/cmd/mesh-enrolment/end.go index 7d2584e..1a00b8b 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/end.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/end.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "errors" "fmt" "strings" @@ -86,25 +87,83 @@ func (h *Holder) Heard(op, key string, value []byte) error { case stdio.OpExpire: e, known := h.seen[node] delete(h.seen, node) - if !known { - // The key expired before the watch handed its content over (a holder started after it was written, before its - // catch-up): the enrolment-issued event cannot be named, and an expiry naming none is no expiry (the contract - // refuses it). Nothing is said; the token is dropped at once, as an ending event would have it dropped. - h.logf("the enrolment of %s expired before its content was seen, so enrolment-expired cannot name its issued event and is not said; its token is dropped", node) - if err := h.Vault.Drop(node); err != nil { - return fmt.Errorf("the token of %s, whose enrolment expired unseen, is not dropped (offered again): %w", node, err) + if known { + return h.endExpired(e) + } + // The key expired before the watch handed its content over: its address key still names its ask and issue (ADR 0589 §6). + return h.reconcileLocked() + } + return nil +} + +// Reconcile ends every enrolment whose key is gone while its address key still names its ask: one that expired while the +// holder was down, or before its watch caught up, or whose end stopped midway (ADR 0589 §6). Run at start, after each +// catch-up and on every expiry; a step that fails is tried again on the next pass. +func (h *Holder) Reconcile() error { + h.mu.Lock() + defer h.mu.Unlock() + return h.reconcileLocked() +} + +func (h *Holder) reconcileLocked() error { + keys, err := h.Store.Keys() + if err != nil { + return fmt.Errorf("the enrolments cannot be listed: %w", err) + } + open := map[string]bool{} + for _, k := range keys { + if n, ok := strings.CutPrefix(k, "node."); ok { + open[n] = true + } + } + var errs []error + for _, k := range keys { + address, ok := strings.CutPrefix(k, "address.") + if !ok { + continue + } + row, found, err := h.Store.Get(k) + if err != nil || !found { + continue + } + r, err := delivery.DecodeEnrolmentAddress(row.Value) + // An address with no ask is reserved by an issue under way: its enrolment key is written next, then the ask. + if err != nil || r.Ask == "" || open[r.Node] { + continue + } + h.logf("the enrolment of %s (ask %s) is gone and its address %s is still reserved: it expired unseen, and is ended now", r.Node, r.Ask, address) + if err := h.endExpired(Enrolment{Node: r.Node, Address: address, Ask: r.Ask, Issued: r.Issued}); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +// endExpired ends an enrolment whose key expired, in this order, each step safe to do again: the node, placed on the yes and +// never confirmed, is taken out of placement on the warrant of its ask (only when placement holds it); enrolment-expired names +// its issued event (said under one id, so said again is said once); the vault drops the token; the address key goes last, so +// a step that failed is found and done on the next pass (ADR 0589 §6). +func (h *Holder) endExpired(e Enrolment) error { + if err := h.unplace(e, "expired"); err != nil { + return fmt.Errorf("%s, whose enrolment expired, is not taken out of placement (tried again on the next pass): %w", e.Node, err) + } + body, err := delivery.EncodeEnrolmentExpiry(delivery.EnrolmentExpiry{Node: e.Node, IssuedID: delivery.EnrolmentIssuedID(e.Node, e.Issued)}) + if err != nil { + return err + } + if err := h.say(EventExpired, json.RawMessage(body), eventID(EventExpired, e)); err != nil { + return fmt.Errorf("enrolment-expired of %s cannot be said (tried again on the next pass): %w", e.Node, err) + } + if err := h.Vault.Drop(e.Node); err != nil { + return fmt.Errorf("the token of %s, whose enrolment expired, is not dropped (tried again on the next pass): %w", e.Node, err) + } + if row, found, err := h.Store.Get(AddressKey(e.Address)); err != nil { + return fmt.Errorf("the address key %s cannot be read (tried again on the next pass): %w", e.Address, err) + } else if found { + if r, err := delivery.DecodeEnrolmentAddress(row.Value); err == nil && r.Node == e.Node && r.Ask == e.Ask { + if err := h.Store.Delete(AddressKey(e.Address)); err != nil { + return fmt.Errorf("the address key %s is not deleted (tried again on the next pass): %w", e.Address, err) } - return nil - } - if err := h.unplace(e, "expired"); err != nil { - return fmt.Errorf("%s, whose enrolment expired, is not taken off placement (offered again): %w", node, err) - } - body, err := delivery.EncodeEnrolmentExpiry(delivery.EnrolmentExpiry{Node: node, IssuedID: eventID(EventIssued, e)}) - if err != nil { - return err - } - if err := h.say(EventExpired, json.RawMessage(body), eventID(EventExpired, e)); err != nil { - return fmt.Errorf("enrolment-expired of %s cannot be said (offered again): %w", node, err) } } return nil diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/end_test.go b/modules/mesh-enrolment/cmd/mesh-enrolment/end_test.go index e812414..40f1d71 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/end_test.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/end_test.go @@ -7,6 +7,7 @@ import ( "slices" "strings" "testing" + "time" stdio "git.novox.be/novox/mesh-sdk/go" ) @@ -53,13 +54,48 @@ func revokeEndsAtAnyStageAndAnExpiryMarkerSaysExpired(t *testing.T) { } last := r.said[len(r.said)-1] x, err := delivery.DecodeEnrolmentExpiry(last.Body) - if err != nil || last.Event != EventExpired || x.Node != "novox" || !strings.HasPrefix(x.IssuedID, "enrolment-issued.novox.") { - t.Errorf("said %s %s: %+v, %v", last.Event, last.Body, x, err) + if err != nil || last.Event != EventExpired || x.Node != "novox" || !strings.HasPrefix(x.IssuedID, "enrolment-issued.novox.") || + !slices.Contains(r.vault.calls, "drop novox") || r.has(AddressKey("10.9.0.4")) { + t.Errorf("said %s %s: %+v, %v; vault %v, keys %v", last.Event, last.Body, x, err, r.vault.calls, r.keys()) } - // A key that expired before its content was seen names no issued event: nothing is said, and the token is dropped. - said := len(r.said) - if err := r.h.Heard(stdio.OpExpire, NodeKey("unseen"), nil); err != nil || len(r.said) != said || !slices.Contains(r.vault.calls, "drop unseen") { - t.Errorf("an unseen expiry: %v, said %v, vault %v", err, r.events(), r.vault.calls) + // ADR 0589 §6: an enrolment whose key expired unseen (before the watch handed it over, or while the holder was down) is + // found by its address key, which names its ask and issue: it leaves placement on that ask's warrant, enrolment-expired + // names its rebuilt issued event, the token is dropped and the address key goes last. A step that fails is done again on + // the next pass, the address key kept until then. + for _, c := range []struct { + name string + node string + fail string + passS func(r *rig) error + }{ + {"expired unseen", "unseen", "", func(r *rig) error { return r.h.Heard(stdio.OpExpire, NodeKey("unseen"), nil) }}, + {"the holder was down, and its drop failed once", "down", "drop down", func(r *rig) error { return r.h.Reconcile() }}, + } { + r.seed(c.node, "10.9.0.9", StageLive) + issued := r.now.Add(-time.Minute) + _ = r.st.Delete(NodeKey(c.node)) // the bus took the key; the watch never handed its content over + if c.fail != "" { + r.vault.fail[c.fail] = true + if err := c.passS(r); err == nil || !r.has(AddressKey("10.9.0.9")) { + t.Errorf("%s: a failed drop ended the pass (%v) or took the address key", c.name, err) + } + } + if err := c.passS(r); err != nil { + t.Fatalf("%s: %v", c.name, err) + } + last := r.said[len(r.said)-1] + x, err := delivery.DecodeEnrolmentExpiry(last.Body) + _, placed := r.rd["mesh-delivery-assigner.placement"][c.node+".node"] + removed := r.added[len(r.added)-1] + if err != nil || x.IssuedID != delivery.EnrolmentIssuedID(c.node, issued) || placed || removed.Change.Op != delivery.OpRemoveNode || + removed.Change.Node != c.node || removed.Warrant.Ask != "ask-"+c.node || !slices.Contains(r.vault.calls, "drop "+c.node) || r.has(AddressKey("10.9.0.9")) { + t.Errorf("%s: said %s (%v), placed %v, last placement call %+v, vault %v, keys %v", c.name, last.Body, err, placed, removed, r.vault.calls, r.keys()) + } + } + // An address reserved by an issue under way (no ask yet) is never taken for an expired enrolment. + _, _ = r.st.Put(AddressKey("10.9.0.10"), Reservation{Node: "busy"}, 0) + if err := r.h.Reconcile(); err != nil || !r.has(AddressKey("10.9.0.10")) { + t.Errorf("a reservation under way was ended: %v", err) } } diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/enrolment.go b/modules/mesh-enrolment/cmd/mesh-enrolment/enrolment.go index f492b5d..9994ad5 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/enrolment.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/enrolment.go @@ -86,7 +86,7 @@ func Encode(v any) (json.RawMessage, error) { case Enrolment: return delivery.EncodeEnrolment(x) case Reservation: - return delivery.EncodeEnrolmentAddress(x.Node) + return delivery.EncodeEnrolmentAddress(x.Node, x.Ask, x.Issued) } return json.Marshal(v) } diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/integration_test.go b/modules/mesh-enrolment/cmd/mesh-enrolment/integration_test.go index ecacd45..b351880 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/integration_test.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/integration_test.go @@ -16,6 +16,7 @@ import ( "slices" "strings" "sync" + "sync/atomic" "testing" "time" @@ -283,6 +284,7 @@ func TestMeshEnrolmentWholeOnARealBus(t *testing.T) { var usersMu sync.Mutex users := map[string]bool{} hasUser := func(u string) bool { usersMu.Lock(); defer usersMu.Unlock(); return users[u] } + var down atomic.Bool // the holder is down: its watch hands nothing over w, err := enrolments.WatchAll(ctx) if err != nil { t.Fatal(err) @@ -294,7 +296,7 @@ func TestMeshEnrolmentWholeOnARealBus(t *testing.T) { continue } node, ok := strings.CutPrefix(e.Key(), "node.") - if !ok { + if !ok || down.Load() { continue } if e.Operation() == jetstream.KeyValuePut { @@ -411,4 +413,40 @@ func TestMeshEnrolmentWholeOnARealBus(t *testing.T) { if _, err := placement.Get(ctx, "g14.node"); err == nil { t.Error("g14 stays in placement after its enrolment expired") } + // Its address key had no expiry of its own: the holder took it last, after the drop (ADR 0589 §6). + waitFor("the expired enrolment's address key is deleted", func() bool { keys, _ := store.Keys(); return len(keys) == 0 }) + + // The holder was down when an enrolment expired: a holder started later finds its address key, which names its ask, and + // ends it the same way (here: the key deleted under the holder, then a pass as at start). + b, err := h.Issue("person.operator", "novox", "a third machine", tunnelKey, 0) + if err != nil { + t.Fatal(err) + } + if err := h.Warranted(asks.Option{ID: "yes"}, asks.Warrant{Ask: b.Ask}); err != nil { + t.Fatal(err) + } + if _, err := placement.Get(ctx, "novox.node"); err != nil { + t.Fatalf("novox was not placed on the yes: %v", err) + } + down.Store(true) + if err := enrolments.Purge(ctx, NodeKey("novox")); err != nil { // the bus took the key while no holder watched + t.Fatal(err) + } + waitFor("the key is gone", func() bool { _, found, _ := store.Get(NodeKey("novox")); return !found }) + if _, err := placement.Get(ctx, "novox.node"); err != nil { + t.Fatalf("novox left placement while no holder ran: %v", err) + } + if err := h.Reconcile(); err != nil { + t.Fatal(err) + } + if _, err := placement.Get(ctx, "novox.node"); err == nil { + t.Error("novox, expired while the holder was down, stays in placement") + } + waitFor("the token of novox is dropped and its address key goes", func() bool { + vault.mu.Lock() + dropped := slices.Contains(vault.calls, "drop novox") + vault.mu.Unlock() + keys, _ := store.Keys() + return dropped && len(keys) == 0 + }) } diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/issue.go b/modules/mesh-enrolment/cmd/mesh-enrolment/issue.go index 1ddc577..e52f3e8 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/issue.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/issue.go @@ -196,6 +196,11 @@ func (h *Holder) raise(e Enrolment) error { if err != nil { return err } + // The address key names the ask and the issue's time from now, with no expiry of its own: it outlives the enrolment key, + // so an enrolment that expires unseen can still be taken out of placement on its ask's warrant (ADR 0589 §6). + if _, err := h.Store.Put(AddressKey(e.Address), Reservation{Node: e.Node, Ask: e.Ask, Issued: e.Issued}, 0); err != nil { + return fmt.Errorf("the address key %s cannot name the ask %s, so the operator is not asked: %w", e.Address, e.Ask, err) + } if err := h.Asker.Ask(ask); err != nil { if cerr := h.Asker.Cancel(ask.ID); cerr != nil { h.logf("the ask %s that could not be published was not taken back at the router: %v", ask.ID, cerr) @@ -265,9 +270,6 @@ func (h *Holder) Warranted(o asks.Option, w asks.Warrant) error { if _, err := h.Store.Put(NodeKey(e.Node), e, ttl); err != nil { return fmt.Errorf("the enrolment of %s cannot be moved to live: %w", e.Node, err) } - if _, err := h.Store.Put(AddressKey(e.Address), Reservation{Node: e.Node}, ttl); err != nil { - return fmt.Errorf("the address key %s cannot be given its expiry: %w", e.Address, err) - } return h.sayEvent(EventIssued, e, "") } diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/issue_test.go b/modules/mesh-enrolment/cmd/mesh-enrolment/issue_test.go index f622fb7..7c59eaf 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/issue_test.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/issue_test.go @@ -99,10 +99,16 @@ func twoIssuesAtOnceHoldTwoAddressesAndAYesMakesTheToken(t *testing.T) { t.Fatal(err) } live, _ := r.enrolment("ace") - if live.Stage != StageLive || r.st.ttl[NodeKey("ace")] != time.Hour || r.st.ttl[AddressKey(live.Address)] != time.Hour || + if live.Stage != StageLive || r.st.ttl[NodeKey("ace")] != time.Hour || r.st.ttl[AddressKey(live.Address)] != 0 || len(r.warr) != 1 || r.warr[0] != e.Ask || r.events()[0] != EventIssued { t.Errorf("live %+v, ttl %v/%v, warrants %v, events %v", live, r.st.ttl[NodeKey("ace")], r.st.ttl[AddressKey(live.Address)], r.warr, r.events()) } + // The address key names the ask and the issue, with no expiry of its own, so it outlives the enrolment key (ADR 0589 §6). + if row, _, _ := r.st.Get(AddressKey(live.Address)); true { + if res, err := delivery.DecodeEnrolmentAddress(row.Value); err != nil || res.Ask != e.Ask || !res.Issued.Equal(live.Issued) { + t.Errorf("the address key %s: %+v, %v", row.Value, res, err) + } + } if got, want := r.vault.tunnels["ace"], (delivery.TokenTunnel{Key: tunnelKey, Address: live.Address + "/32", Range: "10.9.0.0/24", HubKey: "hub-tunnel-key", HubEndpoint: "192.0.2.1:51820"}); got != want { t.Errorf("the token's tunnel %+v, want %+v", got, want) diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/redeem.go b/modules/mesh-enrolment/cmd/mesh-enrolment/redeem.go index 93a8355..7ee0b09 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/redeem.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/redeem.go @@ -116,10 +116,8 @@ func (h *Holder) spend(node, digest, overlay string) (Enrolment, error) { } return Enrolment{}, fmt.Errorf("the enrolment of %s cannot be spent: %w", node, err) } - // The move to redeeming rewrites both keys without an expiry, so a redeem under way never loses its key (§5). - if _, err := h.Store.Put(AddressKey(e.Address), Reservation{Node: node}, 0); err != nil { - return Enrolment{}, fmt.Errorf("the address key %s cannot be kept (call redeem again with the same keys): %w", e.Address, err) - } + // The move to redeeming rewrites the key without an expiry, so a redeem under way never loses it (§5); the address key + // never had one. return e, nil } return Enrolment{}, fmt.Errorf("the enrolment of %s kept changing: call redeem again", node) diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/redeem_test.go b/modules/mesh-enrolment/cmd/mesh-enrolment/redeem_test.go index e8b5b17..6edebd4 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/redeem_test.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/redeem_test.go @@ -81,14 +81,14 @@ func redeemRefusesWhatItMustAndSpendsTheTokenOnce(t *testing.T) { } r := newRig(t) r.seed("ace", "10.9.0.2", StageLive) - r.st.ttl[NodeKey("ace")], r.st.ttl[AddressKey("10.9.0.2")] = time.Hour, time.Hour + r.st.ttl[NodeKey("ace")] = time.Hour ans, err := r.h.Redeem("enrol.ace", good) if err != nil { t.Fatal(err) } e, _ := r.enrolment("ace") - if e.Stage != StageRedeeming || e.KeysDigest != good.KeysDigest() || r.st.ttl[NodeKey("ace")] != 0 || r.st.ttl[AddressKey("10.9.0.2")] != 0 { - t.Errorf("spent %+v, ttl %v/%v", e, r.st.ttl[NodeKey("ace")], r.st.ttl[AddressKey("10.9.0.2")]) + if e.Stage != StageRedeeming || e.KeysDigest != good.KeysDigest() || r.st.ttl[NodeKey("ace")] != 0 { + t.Errorf("spent %+v, ttl %v", e, r.st.ttl[NodeKey("ace")]) } // Redeem writes no placement: the yes did. It binds the identity keys, which the vault checks against the account's digest. if len(r.added) != 0 || ans.Signature == "" || ans.Account != "account-ace" || ans.Login != "node.ace" || ans.Signer != "vault-signer" || ans.Certificate != "PEM-serving-a" { diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/run.go b/modules/mesh-enrolment/cmd/mesh-enrolment/run.go index 07cd094..b23efc9 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/run.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/run.go @@ -10,7 +10,18 @@ import ( // markers), the durable consumer of its own ending events (the drop), and the operator's answers. Each is asked of the // runtime as soon as it serves, and again with a growing wait until the runtime takes it. func (h *Holder) Run(listen func() error) { - go stdio.State(State).Follow("", func(c stdio.StateChange) error { return h.Heard(c.Op, c.Key, c.Value) }) + go func() { + stdio.State(State).Follow("", func(c stdio.StateChange) error { + err := h.Heard(c.Op, c.Key, c.Value) + if err != nil && c.Op == stdio.OpExpire { + // An expiry whose end stopped is finished by a later pass (ADR 0589 §6). + go retry("finish the enrolments that expired", h.logf, h.Reconcile) + } + return err + }) + // Caught up: an enrolment that expired while the holder was down is ended now. + retry("end the enrolments that expired unseen", h.logf, h.Reconcile) + }() for _, event := range EndEvents { go func(key string) { <-stdio.Served() diff --git a/modules/mesh-enrolment/cmd/mesh-enrolment/tokens_test.go b/modules/mesh-enrolment/cmd/mesh-enrolment/tokens_test.go index 00a691c..f7e9c01 100644 --- a/modules/mesh-enrolment/cmd/mesh-enrolment/tokens_test.go +++ b/modules/mesh-enrolment/cmd/mesh-enrolment/tokens_test.go @@ -19,7 +19,7 @@ func (r *rig) seed(node, address string, stage string) { e.KeysDigest = delivery.KeysDigest("pk", "ov", "se", "sv") } _, _ = r.st.Put(NodeKey(node), e, 0) - _, _ = r.st.Put(AddressKey(address), Reservation{Node: node}, 0) + _, _ = r.st.Put(AddressKey(address), Reservation{Node: node, Ask: e.Ask, Issued: e.Issued}, 0) if stage != StageAsked { // the operator's yes placed it, with its tunnel key r.rd["mesh-delivery-assigner.placement"][node+".node"] = `{"v":1,"node":"` + node + `","decided":{"address":"` + address + `"},"modules":[]}` }