modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go

@@ -192,6 +192,7 @@ type delivered interface {
 	Subject() string
 	Data() []byte
 	Ack() error
+	DoubleAck(ctx context.Context) error
 	Term() error
 }
 
@@ -212,7 +213,12 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, logger *log.Log
 	case err != nil:
 		logger.Printf("%s: %s#%d left for redelivery: %v", m.Subject(), f.Repository, f.Number, err)
 	default:
-		_ = m.Ack()
+		// Waits for the server, so the consumer's info no longer counts the message as unacknowledged when the fill that
+		// follows asks whether the stream is applied (issue 1280); a failed ack is redelivered after the ack wait, which
+		// kicks the fill again.
+		if err := m.DoubleAck(ctx); err != nil {
+			logger.Printf("%s: %s#%d applied, its acknowledgement was not confirmed: %v", m.Subject(), f.Repository, f.Number, err)
+		}
 	}
 }
 
@@ -468,7 +474,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		// Essential: the forge's merges are what the indexer indexes.
 		return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the indexer hears no merge without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, forgeStream, err)
 	}
-	cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, ix, logger) },
+	// A merge applied while the taken state was missing, and the gate's heads once the forge stream is applied whole, get
+	// their taken key (issue 1280): when the state appears (Ready runs once, now or then), and after each applied message.
+	ix.Drained = func(ctx context.Context) (bool, error) {
+		info, err := cons.Info(ctx)
+		if err != nil {
+			return false, err
+		}
+		return info.NumPending == 0 && info.NumAckPending == 0, nil
+	}
+	taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) })
+	cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, ix, logger); ix.KickFill(ctx) },
 		jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 	if err != nil {
 		return err

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go

@@ -16,12 +16,18 @@ type fakeMsg struct {
 	subject      string
 	data         []byte
 	acked, termd int
+	doubled      int // acked by DoubleAck, which waits for the server
 }
 
 func (f *fakeMsg) Subject() string { return f.subject }
 func (f *fakeMsg) Data() []byte    { return f.data }
 func (f *fakeMsg) Ack() error      { f.acked++; return nil }
-func (f *fakeMsg) Term() error     { f.termd++; return nil }
+func (f *fakeMsg) DoubleAck(context.Context) error {
+	f.acked++
+	f.doubled++
+	return nil
+}
+func (f *fakeMsg) Term() error { f.termd++; return nil }
 
 type memStore struct{ vals map[string][]byte }
 
@@ -53,7 +59,7 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part)
 	m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull", data: data}
 	handle(context.Background(), m, ix, log.New(&bytes.Buffer{}, "", 0))
 	c, err := delivery.DecodeCandidate(st.vals["feat/x"])
-	if err != nil || len(c.Pulls) != 1 || c.Pulls[0].Modules[0] != "a" || m.acked != 1 || m.termd != 0 {
+	if err != nil || len(c.Pulls) != 1 || c.Pulls[0].Modules[0] != "a" || m.acked != 1 || m.doubled != 1 || m.termd != 0 {
 		t.Errorf("candidate %+v err %v acked %d terminated %d", c, err, m.acked, m.termd)
 	}
 }

modules/mesh-delivery-indexer/internal/index/candidates.go

@@ -69,6 +69,9 @@ type Indexer struct {
 	Placement Store
 	// Say says one of the indexer's events (placement-changed); nil says none.
 	Say func(event string, body any) error
+	// Drained says whether the indexer has applied every message of the forge stream (the consumer has none pending and none
+	// unacknowledged); FillTaken writes the source's heads only then. Nil fills none from the source (taken.go).
+	Drained func(ctx context.Context) (bool, error)
 
 	mu         sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
 	source     sourceNow
@@ -80,6 +83,9 @@ type Indexer struct {
 	retryAgain atomic.Bool                              // an event asked for one more RetryUnread after the running one
 	retryWG    sync.WaitGroup                           // the background RetryUnread, for tests to wait on
 	recompute  sync.Mutex                               // one Recompute at a time
+	untaken    map[string]string                        // repository to the latest merge applied while the taken state was missing (taken.go)
+	filling    atomic.Bool                              // a FillTaken runs in the background (KickFill)
+	fillAgain  atomic.Bool                              // a kick arrived while it ran
 }
 
 func (ix *Indexer) logf(format string, args ...any) {

modules/mesh-delivery-indexer/internal/index/modules_test.go

@@ -119,6 +119,52 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
 				m.Merge = next
 				return ix.Apply(context.Background(), m)
 			}},
+		{name: "a merge heard while the taken state was missing is filled when the state appears", writes: 2, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next,
+			then: func(ix *Indexer) error { // issue 1280: the key was missing, the merge was taken in without it
+				on := ix.Taken.(*memStore)
+				key, _ := delivery.RepositoryKey(repo)
+				delete(on.vals, key)
+				ix.Taken = &memStore{unbound: process.ErrNotYet}
+				m := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"})
+				m.Merge = next
+				if err := ix.Apply(context.Background(), m); err != nil {
+					return err
+				}
+				if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil {
+					return err
+				}
+				ix.Taken = on
+				return ix.FillTaken(context.Background()) // no Drained: only the merge applied is written, not the source's word
+			}},
+		{name: "a source head whose merge was not applied is not a taken key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: "",
+			then: func(ix *Indexer) error { // the forge stream still holds the merge: the consumer is not drained
+				delete(ix.Taken.(*memStore).vals, takenKey(repo))
+				ix.Drained = func(context.Context) (bool, error) { return false, nil }
+				if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil {
+					return err
+				}
+				return ix.FillTaken(context.Background())
+			}},
+		{name: "a source head of a drained forge stream fills a missing key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: next,
+			then: func(ix *Indexer) error {
+				delete(ix.Taken.(*memStore).vals, takenKey(repo))
+				ix.Drained = func(context.Context) (bool, error) { return true, nil }
+				if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil {
+					return err
+				}
+				return ix.FillTaken(context.Background())
+			}},
+		{name: "filling from an older source head leaves a later key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: next,
+			then: func(ix *Indexer) error {
+				ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == merge && in == next, nil }
+				m := mergedMsg(repo, three)
+				m.Merge = next
+				if err := ix.Apply(context.Background(), m); err != nil {
+					return err
+				}
+				ix.Drained = func(context.Context) (bool, error) { return true, nil }
+				return ix.FillTaken(context.Background()) // the source still names merge
+			}},
 		{name: "an earlier merge heard again after a later one leaves taken", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge,
 			then: func(ix *Indexer) error {
 				ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == next && in == merge, nil }
@@ -155,6 +201,9 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
 				}
 			}
 			key, _ := delivery.RepositoryKey(repo)
+			if _, held := taken.vals[key]; !held && tc.taken == "" {
+				return
+			}
 			if got, err := delivery.DecodeTaken(taken.vals[key]); err != nil || got.Merge != tc.taken || got.Repository != repo {
 				t.Errorf("taken.%s = %+v (%v), want merge %.8s", key, got, err, tc.taken)
 			}
@@ -281,3 +330,8 @@ func TestTwoDirectoriesNamingOneModuleWriteOneAndSayTheOther(t *testing.T) { //
 		t.Errorf("module = %+v, want the held directory kept", m)
 	}
 }
+
+func takenKey(repo string) string {
+	key, _ := delivery.RepositoryKey(repo)
+	return key
+}

modules/mesh-delivery-indexer/internal/index/taken.go

@@ -3,6 +3,9 @@ package index
 import (
 	"context"
 	"errors"
+	"fmt"
+	"maps"
+	"slices"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -52,9 +55,98 @@ func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error {
 	})
 	if errors.Is(err, process.ErrNotYet) {
 		// The state is not on the bus yet (a delivery that adds it, bound when it appears as issue 1790 says): said, and
-		// the merge's modules are still taken in; the next merge of the repository writes its key once the state appears.
-		ix.logf("%s at %s: its taken key is not written: %v", f.Repository, f.Merge, err)
+		// the merge's modules are still taken in; FillTaken writes its key once the state appears.
+		ix.logf("%s at %s: its taken key is not written yet, FillTaken writes it when the state appears: %v", f.Repository, f.Merge, err)
+		ix.mu.Lock()
+		if ix.untaken == nil {
+			ix.untaken = map[string]string{}
+		}
+		ix.untaken[f.Repository] = f.Merge // the latest heard; FillTaken keeps a later key over an earlier merge
+		ix.mu.Unlock()
 		return nil
 	}
 	return err
 }
+
+// FillTaken writes the taken keys that were left unwritten, never a merge the indexer has not applied whole (a key names
+// a merge only after it was applied: ADR 0556 clause 1, and the view reads it as the upper bound of what is reached). Two
+// sources, both through writeTaken as a live merge does (issue 1280):
+//   - the merges applied while the state was not on the bus (kept by writeTaken), whenever the state is there;
+//   - the gate's source heads, only when Drained says the indexer has applied every message of the forge stream that was
+//     there before the heads were read. Otherwise it says so, and the next kick of the fill (after a message is applied)
+//     tries again.
+//
+// It is idempotent: a key naming the merge or a later one is left. A state not on the bus yet writes nothing and says
+// nothing (the fill runs again when it appears). The first error is returned after every repository was tried.
+func (ix *Indexer) FillTaken(ctx context.Context) error {
+	if ix.Taken == nil {
+		return nil
+	}
+	if _, err := ix.Taken.Keys(ctx); errors.Is(err, process.ErrNotYet) {
+		return nil
+	}
+	ix.mu.Lock()
+	pending := maps.Clone(ix.untaken)
+	heads := maps.Clone(ix.source.heads)
+	ix.mu.Unlock()
+	var first error
+	note := func(repository string, err error) {
+		if err != nil && first == nil {
+			first = fmt.Errorf("%s: %w", repository, err)
+		}
+	}
+	for _, repository := range slices.Sorted(maps.Keys(pending)) {
+		err := ix.writeTaken(ctx, delivery.Forge{Repository: repository, Merge: pending[repository]})
+		if err == nil {
+			ix.mu.Lock()
+			if ix.untaken[repository] == pending[repository] {
+				delete(ix.untaken, repository)
+			}
+			ix.mu.Unlock()
+		}
+		note(repository, err)
+	}
+	if ix.Drained == nil {
+		return first
+	}
+	if len(heads) == 0 {
+		ix.logf("the source names no repository head, so no taken key is filled from it")
+		return first
+	}
+	done, err := ix.Drained(ctx)
+	if err != nil {
+		ix.logf("whether the forge stream is applied cannot be read, so no taken key is filled from the source: %v", err)
+		return first
+	}
+	if !done {
+		ix.logf("the forge stream is not applied whole yet, so no taken key is filled from the source; the next applied message tries again")
+		return first
+	}
+	for _, repository := range slices.Sorted(maps.Keys(heads)) {
+		note(repository, ix.writeTaken(ctx, delivery.Forge{Repository: repository, Merge: heads[repository]}))
+	}
+	return first
+}
+
+// KickFill runs FillTaken in the background after a message was applied, one at a time; a kick that arrives while one runs
+// makes it run once more.
+func (ix *Indexer) KickFill(ctx context.Context) {
+	if ix.filling.Swap(true) {
+		ix.fillAgain.Store(true)
+		return
+	}
+	go func() {
+		for {
+			ix.fillAgain.Store(false)
+			if err := ix.FillTaken(ctx); err != nil {
+				ix.logf("filling the taken keys: %v", err)
+			}
+			if !ix.fillAgain.Load() {
+				ix.filling.Store(false)
+				if !ix.fillAgain.Load() || ix.filling.Swap(true) {
+					return
+				}
+			}
+		}
+	}()
+}