mesh-enrolment: end an enrolment that expired unseen from its address key, taking the node out of placement

ADR 0589 clause 6: a placed but unconfirmed node is always taken out again, on the warrant of its ask. The address key now
names the ask and the issue's time (delivery.EnrolmentAddress, mesh-sdk c7130e3), written just before the ask is raised and
with no expiry of its own, so it outlives the enrolment key; the holder deletes it at every end. An expiry, seen or not, ends
the enrolment in this order, each step safe to repeat: remove-node on the ask's warrant (only when placement holds the
node), enrolment-expired naming the rebuilt issued event (delivery.EnrolmentIssuedID), the vault's drop of the token, and
the address key last. Reconcile finds every address key that names an ask and whose enrolment key is gone (expired while
the holder was down, before its watch caught up, or an end that stopped) and ends it; it runs once the watch has caught up,
on every expiry, and again with a growing wait while a step fails. An address reserved by an issue under way (no ask yet) is
left alone. Tests: rows in the end table (expired unseen; the holder down with a drop that fails once; a reservation under
way) and the integration test's expiry case (the address key goes; a holder that was down unplaces on its pass); no new
function.

(issue 774)

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL

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

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)
 	}
 }
 

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)
 }

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
+	})
 }

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, "")
 }
 

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)

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)

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" {

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()

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":[]}`
 	}