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

Taken keys are filled only from merges the indexer applied or a drained forge stream's heads (issue 1280)

Browse Source
This commit is contained in:
jschoubben
2026-10-27 00:18:23 +01:00
parent 73bf290f58
commit 0b980203fe
4 changed files with 125 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+11 -10
View File
@@ -363,15 +363,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
go func() { sourceErrs <- follow(ctx, sw, each) }()
}
}
// Every repository the source names gets its taken key at start, and again when the state appears, from the source's
// head and through the path a merge uses (issue 1280): a merge heard before the state existed wrote none.
taken.Ready(func(jetstream.KeyValue) {
go func() {
if err := ix.FillTaken(ctx); err != nil {
logger.Printf("filling the taken keys from the source: %v", err)
}
}()
})
// Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's
// assignments are settled first, so the toolchains settle through the gate already knowing them.
gate := newToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf)
@@ -477,7 +468,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/internal/index/candidates.go
+6
View File
@@ -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
+28 -1
View File
@@ -119,7 +119,7 @@ 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 from the source when it appears", writes: 2, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next,
{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)
@@ -134,6 +134,24 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
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,
@@ -144,6 +162,7 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
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,
@@ -182,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)
}
@@ -308,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
+80 -9
View File
@@ -55,27 +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 key of every repository the gate's source names, at the head it names, through writeTaken as a
// live merge does (issue 1280): a merge heard while the state was not on the bus left its repository without a key until
// that repository's next merge. It runs at start and when the state appears. A repository whose key names that head or a
// later one is left; a repository the source does not name is left; the first error is returned after every repository
// was tried.
// 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
for _, repository := range slices.Sorted(maps.Keys(heads)) {
if err := ix.writeTaken(ctx, delivery.Forge{Repository: repository, Merge: heads[repository]}); err != nil && first == nil {
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
}
}
}
}()
}