Merge branch 'fix/1945-take-in-consumer' into fix/1945-kept-compositions-follow-config

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