novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

The indexer's taken key moves only by a merge it applied, never by the gate's source head (issue 1126)

Browse Source 
A drained forge stream does not say a merge was applied: the forge bridge was still
retrying mesh-catalog#535's merged message when the indexer started, and the source's
head moved taken.mesh-catalog to 9fd9569, so the view and show said the merge was on the
machines while mesh-vault's record stood at 78d7ef7 (ADR 0556 clause 1).
This commit is contained in:
jschoubben
2026-10-28 11:10:09 +01:00
parent 93c0ae19dd
commit 458e403a04
4 changed files with 23 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+2 -9
View File
@@ -404,15 +404,8 @@ 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)
}
// 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
}
// A merge applied while the taken state was missing gets its taken key (issue 1280): when the state appears (Ready runs
// once, now or then), and after each applied message. The gate's source heads never move a key (issue 1126).
taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) })
cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, logger); ix.KickFill(ctx) }, logger.Printf)
if err != nil {
modules/mesh-delivery-indexer/internal/index/candidates.go
-3
View File
@@ -79,9 +79,6 @@ type Indexer struct {
ReadOpen func(ctx context.Context) ([]delivery.OpenPull, error)
// 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
modules/mesh-delivery-indexer/internal/index/modules_test.go
+14 -16
View File
@@ -139,31 +139,29 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
{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,
// Issue 1126 (mesh-catalog#535): the forge bridge never wrote the merged message of 9fd9569, so the forge stream was
// drained with the merge not applied; the source head moved the key all the same, and the view said the merge was
// on the machines. A key moves only by a merge applied (ADR 0556 clause 1), never by the source's word.
{name: "a source head of a drained forge stream whose merge was never heard leaves the key", 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 == merge && in == next, nil }
m := mergedMsg(repo, three)
m.Merge = next
if err := ix.Apply(context.Background(), m); err != nil {
if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil {
return err
}
ix.Drained = func(context.Context) (bool, error) { return true, nil }
return ix.FillTaken(context.Background()) // the source still names merge
return ix.FillTaken(context.Background())
}},
{name: "a source head of a drained forge stream fills no missing key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: "",
then: func(ix *Indexer) error {
delete(ix.Taken.(*memStore).vals, takenKey(repo))
if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil {
return err
}
return ix.FillTaken(context.Background())
}},
{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 {
modules/mesh-delivery-indexer/internal/index/taken.go
+7 -27
View File
@@ -68,13 +68,13 @@ func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error {
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.
// FillTaken writes the taken keys of the merges applied while the state was not on the bus (kept by writeTaken), through
// writeTaken as a live merge does (issue 1280), whenever the state is there. It never writes 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). The gate's source heads are not such merges: a drained forge stream does not say a merge was applied,
// since the forge bridge may not have written its merged message yet (issue 1126: mesh-catalog#535's merged message was
// still being retried by the bridge when the indexer started, and the source's head moved the key, so the view said the
// merge was on the machines while the indexer held mesh-vault at the merge before).
//
// 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.
@@ -87,7 +87,6 @@ func (ix *Indexer) FillTaken(ctx context.Context) error {
}
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) {
@@ -106,25 +105,6 @@ func (ix *Indexer) FillTaken(ctx context.Context) error {
}
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
}