Say again what was last applied when the mesh asks (hq to-be 45 Phase 3, the report verb)

The controller's healer H1 answers a send that went unreported by asking the
machine first: a report lost on its way (issue 264) needs no second send. The
node-engine now hears mesh.node.<self>.ask.report on core NATS and enqueues a
reconcile whose account is said whether or not it is news; a delivery waiting
meanwhile is applied and reported instead. The answer is an ordinary report on
its own subject, so the node publishes nothing new and answers nobody's inbox.

The genesis lock grants the controller the healer-acted seat event, which it
now composes; the genesis test in mesh-controller holds the two equal.

internal/link/asked_test.go

@@ -0,0 +1,130 @@
+package link
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+)
+
+// The `report` verb (novox/hq to-be 45 §6), what healer H1 asks a machine whose send went unreported:
+// the node says again what it last applied, as an ordinary report through the one queue.
+
+// **Asked, the reconcile's account is said even when it is not news** — the mesh asked because it does
+// not know, and silence would be the very fault it is asking about.
+func TestAskedToReportSaysWhatWasLastAppliedEvenWhenNotNews(t *testing.T) {
+	m, _ := aMember(t)
+	l := newQuietLink()
+	var lines []string
+	q := aQueue(m, (&applies{}).apply, &keptInMemory{})
+	q.Say = func(s string) { lines = append(lines, s) }
+	q.Reconcile = func(context.Context) (Report, bool) {
+		return Report{Declared: "d-kept", Order: Order{Epoch: 7, Sequence: 12}, Applied: []string{"a"}}, true
+	}
+	q.News = func(Report) bool { return false }
+	q.attach(context.Background(), l)
+	ctx, stop := context.WithCancel(context.Background())
+	worker := runWorker(ctx, q)
+
+	q.ReconcileDue() // an ordinary reconcile: not news, not said
+	time.Sleep(50 * time.Millisecond)
+	if n := len(l.said()); n != 0 {
+		t.Fatalf("an ordinary reconcile that is not news said %d report(s)", n)
+	}
+	q.ReportAsked()
+	eventually(t, "asked, the node said nothing", func() bool { return len(l.said()) == 1 })
+	stop()
+	<-worker
+	r := l.said()[0]
+	if r.Declared != "d-kept" || r.Sequence != 12 || r.Epoch != 7 || r.ReportSequence == 0 || r.Node != m.Node {
+		t.Fatalf("the account said is not the one kept, in order: %+v", r)
+	}
+	if !strings.Contains(strings.Join(lines, "\n"), "asked by the mesh") {
+		t.Errorf("being asked is not said in the journal: %q", lines)
+	}
+}
+
+// **A delivery waiting answers the question better**: it is applied, once, and its report is the answer.
+func TestAskedWhileADeliveryWaitsIsThatDeliverysReport(t *testing.T) {
+	m, key := aMember(t)
+	l := newQuietLink()
+	a := &applies{}
+	q := aQueue(m, a.apply, &keptInMemory{})
+	reconciled := 0
+	q.Reconcile = func(context.Context) (Report, bool) { reconciled++; return Report{}, true }
+	q.attach(context.Background(), l)
+	q.Deliver(ordered(t, key, "new", 7, 13))
+	q.ReportAsked()
+	ctx, stop := context.WithCancel(context.Background())
+	worker := runWorker(ctx, q)
+	eventually(t, "the delivery was never reported", func() bool { return len(l.said()) == 1 })
+	time.Sleep(50 * time.Millisecond)
+	stop()
+	<-worker
+	if reconciled != 0 || len(l.said()) != 1 || l.said()[0].Sequence != 13 {
+		t.Fatalf("reconciled %d, said %+v", reconciled, l.said())
+	}
+}
+
+// askingLink is a quiet link the mesh can ask.
+type askingLink struct {
+	*quietLink
+	asks chan struct{}
+}
+
+func (l askingLink) AskedToReport() <-chan struct{} { return l.asks }
+
+// **The link hands the question to the queue**: asked on the link, the account is said on it.
+func TestALinkThatIsAskedHandsItToTheQueue(t *testing.T) {
+	m, _ := aMember(t)
+	l := askingLink{quietLink: newQuietLink(), asks: make(chan struct{}, 1)}
+	q := aQueue(m, (&applies{}).apply, &keptInMemory{})
+	q.Reconcile = func(context.Context) (Report, bool) { return Report{Declared: "d-kept"}, true }
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	worker := runWorker(ctx, q)
+	served := make(chan error, 1)
+	go func() { served <- serve(ctx, l, m, q, nil, time.Second) }()
+	l.asks <- struct{}{}
+	eventually(t, "asked on the link, nothing was said", func() bool {
+		for _, r := range l.said() {
+			if r.Declared == "d-kept" {
+				return true
+			}
+		}
+		return false
+	})
+	stop()
+	<-served
+	<-worker
+}
+
+// **Against a real server**: the question on the node's own subject reaches it, and another's does not.
+func TestNatsTheMeshAsksANodeToReport(t *testing.T) {
+	conn, js := aBus(t)
+	l := &natsLink{conn: conn, js: js, node: "asked"}
+	l.hearAsks()
+	defer l.Close()
+	if err := conn.Publish(AskReportSubject("another"), []byte(`{}`)); err != nil {
+		t.Fatal(err)
+	}
+	if err := conn.Publish(AskReportSubject("asked"), []byte(`{"by":"the controller's healer H1"}`)); err != nil {
+		t.Fatal(err)
+	}
+	if err := conn.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-l.AskedToReport():
+	case <-time.After(5 * time.Second):
+		t.Fatal("the question on the node's own subject never reached it")
+	}
+	select {
+	case <-l.AskedToReport():
+		t.Fatal("asked twice: another node's question reached this one")
+	case <-time.After(200 * time.Millisecond):
+	}
+	if AskReportSubject("asked") != "mesh.node.asked.ask.report" {
+		t.Fatalf("the subject %q is not the one the controller asks on and grants", AskReportSubject("asked"))
+	}
+}

internal/link/hearing.go

@@ -38,6 +38,13 @@ type Link interface {
 	Close()
 }
 
+// Asked is a link that hears the mesh asking this node to say again what it last applied (novox/hq
+// to-be 45 §6, the `report` verb). Optional: a link without it is never asked, and the mesh then sends
+// the current declaration again instead.
+type Asked interface {
+	AskedToReport() <-chan struct{}
+}
+
 // Declaration is one thing the mesh told this node to be.
 //
 // **Handled, once — after the report is published.** A node that dies between applying and

internal/link/hearing_nats.go

@@ -38,6 +38,11 @@ const EnrolSubject = "mesh.control.enrol"
 // for.
 func DeclareSubject(node string) string { return "mesh.node." + node + ".declare" }
 
+// AskReportSubject is where the mesh asks this node to say again what it last applied (novox/hq to-be 45
+// §6): its own, on core NATS and off any stream. The answer is an ordinary report on its report subject,
+// the one thing it already says — nobody's inbox is answered.
+func AskReportSubject(node string) string { return "mesh.node." + node + ".ask.report" }
+
 // natsURL is a bus address as the client wants it. A membership records host and port, because that
 // is what genesis sealed into it and what the other transport takes; the scheme is this transport's
 // own business.
@@ -53,6 +58,8 @@ type natsLink struct {
 	conn    *nats.Conn
 	js      nats.JetStreamContext
 	sub     *nats.Subscription
+	asking  *nats.Subscription
+	asked   chan struct{}
 	node    string
 	arrived chan Declaration
 	lost    chan error
@@ -106,6 +113,7 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e
 		conn: conn, js: js, node: m.Node,
 		arrived: make(chan Declaration, drainDepth),
 		lost:    make(chan error, 1),
+		asked:   make(chan struct{}, 1),
 	}
 
 	// Bound to the consumer the controller made for this node, named after the node because that is
@@ -124,6 +132,8 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e
 	}
 	l.sub = sub
 
+	l.hearAsks()
+
 	conn.SetDisconnectErrHandler(func(_ *nats.Conn, err error) {
 		select {
 		case l.lost <- fmt.Errorf("the link dropped: %w", err):
@@ -163,13 +173,36 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e
 	return l, nil
 }
 
+// hearAsks listens for the mesh asking what this node last applied (the `report` verb). One question
+// waiting is enough: a second asked before the first is taken is the same question. A bus whose user
+// list is older than the grant refuses the subscription asynchronously — said by the client, and the
+// mesh then sends again instead of asking; nothing here depends on being asked.
+func (l *natsLink) hearAsks() {
+	if l.asked == nil {
+		l.asked = make(chan struct{}, 1)
+	}
+	asking, err := l.conn.Subscribe(AskReportSubject(l.node), func(*nats.Msg) {
+		select {
+		case l.asked <- struct{}{}:
+		default:
+		}
+	})
+	if err == nil {
+		l.asking = asking
+	}
+}
+
 func (l *natsLink) Declarations() <-chan Declaration { return l.arrived }
+func (l *natsLink) AskedToReport() <-chan struct{}   { return l.asked }
 func (l *natsLink) Lost() <-chan error               { return l.lost }
 
 func (l *natsLink) Close() {
 	if l.sub != nil {
 		_ = l.sub.Unsubscribe()
 	}
+	if l.asking != nil {
+		_ = l.asking.Unsubscribe()
+	}
 	if l.conn != nil {
 		l.conn.Close()
 	}

internal/link/queue.go

@@ -64,8 +64,11 @@ type Queue struct {
 	delivering bool
 	waiting    []Declaration // delivered and not yet taken, in the order they arrived
 	due        bool          // a reconcile asked for
-	bus        Bus           // the link open now; nil while there is none
-	unasked    *Report       // a reconcile's report not yet said, made while there was no link
+	// asked is the mesh asking this node to say what it last applied (the `report` verb, to-be 45 §6):
+	// the reconcile it enqueues says its report whether or not it is news.
+	asked   bool
+	bus     Bus     // the link open now; nil while there is none
+	unasked *Report // a reconcile's report not yet said, made while there was no link
 
 	// saying is held while a report is made and said, so the order they leave in is the order they
 	// are made in — whoever says them, the worker or a link opening.
@@ -116,6 +119,19 @@ func (q *Queue) ReconcileDue() {
 	q.signal()
 }
 
+// ReportAsked is the mesh asking this node to say again what it last applied (novox/hq to-be 45 §6, the
+// `report` verb its healer H1 asks when a send went unreported). **An enqueue, like everything else**: a
+// reconcile, whose report — the account of the declaration this node keeps, the last it applied — is
+// said whether or not it is news. A delivery waiting meanwhile is applied instead and reported, which
+// answers the question better. Asked again before the worker takes it is the same question.
+func (q *Queue) ReportAsked() {
+	q.init()
+	q.mu.Lock()
+	q.due, q.asked = true, true
+	q.mu.Unlock()
+	q.signal()
+}
+
 func (q *Queue) signal() {
 	select {
 	case q.wake <- struct{}{}:
@@ -147,8 +163,8 @@ func (q *Queue) step(ctx context.Context) bool {
 		q.mu.Unlock()
 		return false
 	}
-	batch, due := q.waiting, q.due
-	q.waiting, q.due = nil, false
+	batch, due, asked := q.waiting, q.due, q.asked
+	q.waiting, q.due, q.asked = nil, false, false
 	q.delivering = len(batch) > 0
 	q.mu.Unlock()
 	defer func() {
@@ -159,7 +175,7 @@ func (q *Queue) step(ctx context.Context) bool {
 	}()
 
 	if len(batch) == 0 {
-		q.reconcile(ctx)
+		q.reconcile(ctx, asked)
 		return true
 	}
 
@@ -210,14 +226,27 @@ func (q *Queue) step(ctx context.Context) bool {
 	return true
 }
 
-func (q *Queue) reconcile(ctx context.Context) {
+func (q *Queue) reconcile(ctx context.Context, asked bool) {
 	if q.Reconcile == nil {
+		if asked {
+			q.Say("asked to say what this node last applied, and it holds nothing to reconcile with")
+		}
 		return
 	}
 	report, ok := q.Reconcile(ctx)
-	if !ok || q.News == nil || !q.News(report) {
+	if !ok {
+		if asked {
+			q.Say("asked to say what this node last applied, and the reconcile could not say it")
+		}
 		return
 	}
+	if !asked && (q.News == nil || !q.News(report)) {
+		return
+	}
+	if asked {
+		q.Say(fmt.Sprintf("asked by the mesh, saying again what this node last applied: declaration %s (%s)",
+			short(report.Declared), report.Order.Words()))
+	}
 	q.tell(ctx, report, unasked)
 }
 

internal/link/run.go

@@ -234,6 +234,11 @@ func serve(ctx context.Context, link Link, m Membership, queue *Queue, say Annou
 	defer queue.detach(link)
 
 	declarations := link.Declarations()
+	// And the mesh asking what this node last applied (to-be 45 §6), on a link that hears the question.
+	var asks <-chan struct{}
+	if a, ok := link.(Asked); ok {
+		asks = a.AskedToReport()
+	}
 
 	for {
 		// Asked to stop — by standing aside, say — nothing further is taken in.
@@ -247,6 +252,8 @@ func serve(ctx context.Context, link Link, m Membership, queue *Queue, say Annou
 			publishAlive(ctx, link, m, say, timeout)
 		case reason := <-link.Lost():
 			return reason
+		case <-asks:
+			queue.ReportAsked()
 		case declaration, ok := <-declarations:
 			if !ok {
 				// The link's own reason, when it has managed to say one: "stopped delivering" on