Register on the fingerprint, not the commit; count the take-in's attempts as settled (issue 1945, ADR 0544, review 3)

registeredAt compared commits, so a catalogue merge that moved only the commit
(2df05030 at 18:02, a85d9037 before it) took every catalogue module in again
and composed it: about 200 take-ins, some 25 minutes, on every mesh-catalog
merge, which held nextcloud for an hour. A module whose trunk fingerprint is
the one it is registered at now has the new commit recorded (built_from and
source_head, as a registration moves them) and composes nothing unless its
composition or grant is missing. Readers of the registered commit: the
fingerprints read the commit's build or, with none, the indexer's value at it,
which names the same fingerprint; status, modules, the board and the facts
read it as the source's commit, which it is; the upgrade path counts it as
built from that commit, which it is.

The stall watch counted only settles the bus accepted, so each refused
acknowledgement at a first start left work held for good: it now counts the
loop's every attempt. The acknowledgement and refusals at the stop are sent
past the stop, and the connection closes only after the readers and the loop
have ended.
Research 072: the take-ins on a bounded pull consumer.

cmd/mesh-controller/composed_keys.go

@@ -40,8 +40,11 @@ type takeInSteps struct {
 	catalogue func(ctx context.Context) (map[string][]string, error)
 	// trunk is the indexer's trunk value of the module.
 	trunk func(ctx context.Context, module string) (sdkdelivery.Module, error)
-	// registeredAt says the module is registered at the commit.
-	registeredAt func(ctx context.Context, module, commit string) (bool, error)
+	// registered says whether the module is registered at the trunk's commit, and whether at its fingerprint: a commit
+	// that moves while the fingerprint stays (a catalogue merge that leaves the module's sources alone) is no new build.
+	registered func(ctx context.Context, module string, trunk sdkdelivery.Module) (atCommit, atFP bool, err error)
+	// recordCommit records the module's registration at the trunk's commit, its fingerprint unchanged: nothing composed.
+	recordCommit func(ctx context.Context, module, commit string) error
 	// takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey).
 	takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error
 	// taken says the module's compositions at fp exist for every node and its grant is written: nothing to compose.
@@ -151,7 +154,14 @@ func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[strin
 			out = append(out, r)
 			continue
 		}
-		at, err := s.registeredAt(ctx, m, trunk.Commit)
+		at, atFP, err := s.registered(ctx, m, trunk)
+		if err == nil && !at && atFP {
+			// Only the commit moved (issue 1945: every catalogue merge took about 200 modules in again, 25 minutes): the
+			// new commit is recorded, and the module is judged as one registered, composed only if something is missing.
+			if err = s.recordCommit(ctx, m, trunk.Commit); err == nil {
+				at = true
+			}
+		}
 		if err != nil {
 			r.verdict, r.err = refused, err
 			out = append(out, r)
@@ -311,12 +321,19 @@ func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) ta
 		trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) {
 			return keys.ModuleAt(ctx, module, "")
 		},
-		registeredAt: func(ctx context.Context, module, commit string) (bool, error) {
+		registered: func(ctx context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) {
 			src, err := inv.SourceOf(ctx, module)
 			if err != nil {
-				return false, err
+				return false, false, err
 			}
-			return sameCommitID(src.BuiltFrom, commit), nil
+			if sameCommitID(src.BuiltFrom, trunk.Commit) {
+				return true, true, nil
+			}
+			fp, err := registeredFingerprint(ctx, inv, keys, module, src.BuiltFrom)
+			return false, err == nil && fp != "" && fp == trunk.FP2, nil
+		},
+		recordCommit: func(ctx context.Context, module, commit string) error {
+			return inv.RecordBuiltFrom(ctx, module, commit)
 		},
 		takeIn: func(ctx context.Context, module string, trunk sdkdelivery.Module) error {
 			_, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: trunk, Ask: registrationAsk{Merge: trunk.Commit}})
@@ -343,6 +360,24 @@ func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) ta
 	}
 }
 
+// registeredFingerprint is the fingerprint a module is registered at, as currentFingerprints reads it: its commit's
+// newest build's, or the indexer's value at the commit when the commit has no build of its own; "" when neither says.
+func registeredFingerprint(ctx context.Context, inv *inventory.Inventory, keys registerKeys, module, commit string) (string, error) {
+	if commit == "" {
+		return "", nil
+	}
+	if id, err := inv.BuildOf(ctx, module, commit); err == nil && id != "" {
+		if b, found, err := inv.BuildByID(ctx, id); err == nil && found && b.SourceFingerprint != "" {
+			return b.SourceFingerprint, nil
+		}
+	}
+	at, err := keys.ModuleAt(ctx, module, commit)
+	if err != nil {
+		return "", err
+	}
+	return at.FP2, nil
+}
+
 // composedAndGranted says the bus holds module's grant at fp and its composition at fp for every node: a few reads of
 // keys, never a composition.
 func composedAndGranted(ctx context.Context, js jetstream.JetStream, module, fp string, nodes []string) (bool, error) {

cmd/mesh-controller/composed_keys_test.go

@@ -19,6 +19,7 @@ import (
 // nodes, whether its compositions and grant exist, and what its take-in answers; every act is written to acts.
 type aTakeIn struct {
 	registered map[string]string
+	sameFP     map[string]bool   // registered at another commit, at the trunk's fingerprint
 	trunks     map[string]string // module -> commit; no entry: no trunk value
 	nodes      map[string][]string
 	taken      map[string]bool
@@ -47,8 +48,15 @@ func (m *aTakeIn) steps() takeInSteps {
 			}
 			return sdkdelivery.Module{Commit: commit, FP2: fp(module)}, nil
 		},
-		registeredAt: func(_ context.Context, module, commit string) (bool, error) {
-			return m.registered[module] == commit, nil
+		registered: func(_ context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) {
+			if m.registered[module] == trunk.Commit {
+				return true, true, nil
+			}
+			return false, m.sameFP[module], nil
+		},
+		recordCommit: func(_ context.Context, module, commit string) error {
+			m.acts = append(m.acts, "record "+module+" at "+commit)
+			return nil
 		},
 		takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error {
 			m.acts = append(m.acts, "take in "+module+" at "+trunk.Commit)
@@ -113,6 +121,12 @@ func TestATakeInBatch(t *testing.T) {
 			keys: []delivered{{module: "mail"}, {module: "nats"}, {module: "mail"}},
 			acts: []string{"catalogue"},
 			want: map[string]want{"mail": {nothingNew: true}, "nats": {nothingNew: true}}},
+		{name: "a catalogue merge that moved the commit and not the fingerprint: recorded, nothing composed",
+			mesh: aTakeIn{registered: map[string]string{"mail": "a85d9037"}, sameFP: map[string]bool{"mail": true},
+				trunks: map[string]string{"mail": "2df05030"}, nodes: map[string][]string{"mail": {"ace"}},
+				taken: map[string]bool{"mail": true}, holder: "novox"},
+			keys: []delivered{{module: "mail"}},
+			acts: []string{"catalogue", "record mail at 2df05030"}, want: map[string]want{"mail": {nothingNew: true}}},
 		{name: "taken in by this process at the same trunk: nothing new, not even a look at its keys",
 			mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"},
 				nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"},
@@ -270,6 +284,10 @@ func TestATakeInStallsWhenItHoldsWorkAndSettlesNone(t *testing.T) {
 			{0, 0, 3, 40, 40, false, true},
 			{30 * time.Minute, 0, 3, 40, 40, false, false},
 		}},
+		{"every key settled, one of them refused by the bus, nothing pending: no stall", []reading{
+			{0, 0, 1, 3, 3, false, true},
+			{6 * time.Minute, 0, 1, 3, 3, false, false},
+		}},
 		{"a batch handed and never settled is a stall", []reading{
 			{0, 0, 16, 56, 40, false, true},
 			{5 * time.Minute, 0, 16, 56, 40, true, true},

cmd/mesh-controller/take_in_consumer.go

@@ -133,6 +133,9 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 	// it): what tells a take-in that moves from one that holds work and does nothing (the stall watch, ADR 0544 §6).
 	moving := &takeInMoving{}
 	read := make([]jetstream.Consumer, 0, len(consumers))
+	// The connection closes only once the readers and the loop have ended, so the refusals they send at the stop reach
+	// the bus (the third review of ADR 0544).
+	var users sync.WaitGroup
 	for _, c := range consumers {
 		bucket := c.bucket
 		read = append(read, c.consumer)
@@ -145,7 +148,9 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 			<-ctx.Done()
 			it.Drain() // the keys it holds are handed out and refused at once below, not lost to their ack wait
 		})
+		users.Add(1)
 		goServing(ctx, func() {
+			defer users.Done()
 			for {
 				m, err := it.Next()
 				if err != nil {
@@ -159,16 +164,20 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 					_ = m.Ack()
 					continue
 				}
+				// Settled is the loop's attempt, whatever the bus answers: an acknowledgement it refuses (at a first start,
+				// until the bus node is composed) is delivered again and settled again, and never counts as work held
+				// (the third review of ADR 0544). Acknowledged synchronously and past the stop, so a refusal is said.
 				msg := takeInMsg{delivered: d, ack: func() {
-					// Counted only when the bus took it: at a first start it refuses the acks until the bus node is composed.
-					if m.DoubleAck(ctx) == nil {
-						moving.settled.Add(1)
+					acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+					defer cancel()
+					if err := m.DoubleAck(acking); err != nil {
+						doneKeysSay(fmt.Sprintf("done keys: the bus did not take the acknowledgement of %s: %v", d.module, err))
 					}
+					moving.settled.Add(1)
 				},
 					nak: func(delay time.Duration) {
-						if m.NakWithDelay(delay) == nil {
-							moving.settled.Add(1)
-						}
+						_ = m.NakWithDelay(delay)
+						moving.settled.Add(1)
 					}}
 				if ctx.Err() != nil {
 					_ = m.Nak()
@@ -186,10 +195,15 @@ func watchDoneKeys(ctx context.Context, open *stores) {
 	}
 	backlog := func(ctx context.Context) (takeInBacklog, error) { return backlogOf(ctx, read) }
 	steps := liveTakeInSteps(open, keys, keys.JS)
-	goServing(ctx, func() { takeInLoop(ctx, in, steps, backlog) })
+	users.Add(1)
+	goServing(ctx, func() {
+		defer users.Done()
+		takeInLoop(ctx, in, steps, backlog)
+	})
 	goServing(ctx, func() { watchTakeInStall(ctx, backlog, moving) })
 	goServing(ctx, func() {
 		<-ctx.Done()
+		users.Wait()
 		closeKeys()
 	})
 }