diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go new file mode 100644 index 0000000..b79ba6b --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -0,0 +1,111 @@ +// Command mesh-delivery-indexer-tools serves the indexer's doing tool (novox/hq issue 1945): the runtime launches it for a +// call. indexer_take_in writes one value of the indexer's takeins state and waits for the indexer's answer there. The +// indexer is the other program: it serves no request subject (ADR 0437 §4). The read tools of the indexer's states are +// mesh-delivery-view's (indexer_module, indexer_modules, indexer_candidate, indexer_candidates). +package main + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + + "mesh-delivery-indexer/internal/index" +) + +// asker writes a takeins value and waits for the indexer's answer to it. +type asker func(key string, value []byte) (index.TakeInAsk, error) + +// Tools are the indexer's tools. +func Tools(ask asker, now func() time.Time) []stdio.Tool { + str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + return []stdio.Tool{ + {Name: index.TakeInTool, + Description: "Take a registered module in now: the indexer reads its manifest from the repository and commit the controller's " + + "build records register it from (at the head the gate's source names, when it names one), whatever it tried before, and " + + "writes it to its modules state as a merge would. For a module registered from outside the indexed repositories whose " + + "repository the forge could not read (a 404 before the forge's account could read it). Answers stored, from which " + + "repository, commit and path, or why not (held already, no registration, the forge's error, no manifest there). " + + "The indexer reads such a module again by itself on the next source revision and merged message; this asks now. " + + "indexer_module (mesh-delivery-view) reads the result.", + Input: map[string]any{"type": "object", "required": []string{"module"}, "properties": map[string]any{ + "module": map[string]any{"type": "string", "description": "the module's name, as the catalogue registers it"}, + "why": map[string]any{"type": "string", "description": "why it is taken in now, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + key, value, err := index.NewTakeIn(str(args, "module"), str(args, "why"), stdio.CallerOf(args), now()) + if err != nil { + return nil, err + } + a, err := ask(key, value) + if err != nil { + return nil, err + } + return a.Result, nil + }}, + } +} + +// askTakeIn writes the takeins value at key and waits, on a watch of the key, at most bound for the indexer's answer. +func askTakeIn(s stdio.KeptState, bound time.Duration) asker { + return func(key string, value []byte) (index.TakeInAsk, error) { + answered := make(chan index.TakeInAsk, 1) + var rev uint64 + var mu sync.Mutex + take := func(revision uint64, v []byte) bool { + mu.Lock() + mine := rev + mu.Unlock() + if mine == 0 || revision <= mine { + return false + } + a, err := index.DecodeTakeIn(v) + if err != nil || a.Answered.IsZero() || a.Result == nil { + return false + } + select { + case answered <- a: + default: + } + return true + } + if err := s.Watch(key, func(c stdio.StateChange) error { + if c.Op == "put" { + take(c.Revision, c.Value) + } + return nil + }); err != nil { + return index.TakeInAsk{}, fmt.Errorf("watching %s.%s: %w", index.TakeInsState, key, err) + } + r, err := s.Put(key, json.RawMessage(value)) + if err != nil { + return index.TakeInAsk{}, fmt.Errorf("the ask could not be written to %s: %w", index.TakeInsBucket, err) + } + mu.Lock() + rev = r + mu.Unlock() + // The watch may have heard the answer before the revision was known: read it once too. + if e, err := s.Get(key); err == nil && e != nil { + take(e.Revision, e.Value) + } + wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8) + defer cancel() + select { + case a := <-answered: + return a, nil + case <-wait.Done(): + return index.TakeInAsk{}, fmt.Errorf("the indexer did not answer within %s: the take-in is still running (a forge read can take 30 s); its answer will be in %s.%s", bound, index.TakeInsBucket, key) + } + } +} + +func main() { + if err := stdio.Serve("", Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now)); err != nil { + fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err) + os.Exit(1) + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 8081a2f..672e4e6 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -1,7 +1,8 @@ // Command mesh-delivery-indexer is delivery's indexer (novox/hq ADR 0437 §4, §5, issue 1238): it hears the forge // bridge's messages on the stream FORGE and keeps the candidate groups (candidates.): the pull requests that share // a branch, their heads and the modules each touches. It reads the controller's builds to know which module a changed -// directory is. It serves no request subject and holds no timer: every reaction is a message or a watch (ADR 0437 §8). +// directory is. It serves no request subject and holds no timer: every reaction is a message or a watch (ADR 0437 §8). Its +// tool indexer_take_in (the program mesh-delivery-indexer-tools) asks through its state takeins, which it watches. // // Its build of 2026-10-24 00:37 named a program whose push never reached the artifact store, and a retry sends that // same build again; this paragraph was added only to give the indexer a new build (issue 1933). @@ -359,6 +360,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }) }() } + // A person takes a registered module in now with indexer_take_in, which writes the takeins state (issue 1945): watched, + // as every other input. A module the forge could not read is also read again by itself on the next source revision and + // merged message (index.RetryUnread). + if kv := process.OptionalBucket(ctx, js, index.TakeInsBucket, "indexer_take_in is not answered until the indexer starts again; a module the forge could not read is still read again by itself", logger.Printf); kv != nil { + eachTakeIn := func(e jetstream.KeyValueEntry) { takenIn(ctx, ix, kv, e, logger.Printf) } + if tw, err := settle(ctx, kv, eachTakeIn); err != nil { + process.NotWatched(logger.Printf, index.TakeInsBucket, err, "indexer_take_in is not answered until the indexer starts again") + } else { + go func() { errs <- follow(ctx, tw, eachTakeIn) }() + } + } // The modules the trunk no longer holds go from the bucket (issue 1900); a merge does the same from its own message. if err := ix.Reconcile(ctx); err != nil { logger.Printf("dropping the modules the trunk does not hold at start: %v", err) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go new file mode 100644 index 0000000..f0c7b02 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -0,0 +1,44 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-delivery-indexer/internal/index" +) + +// takerIn is the part of the indexer a takeins value asks through. +type takerIn interface { + TakeIn(ctx context.Context, module string) index.TakenIn +} + +// takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the +// module in now and is written back, by compare-and-set at its revision, with what the indexer stored or why not. A value +// answered already (this indexer's own write) or deleted is left. +func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream.KeyValueEntry, logf func(string, ...any)) { + if e.Operation() != jetstream.KeyValuePut { + return + } + a, err := index.DecodeTakeIn(e.Value()) + if err != nil { + logf("%s.%s holds what is not a take-in ask, left: %v", index.TakeInsState, e.Key(), err) + return + } + if !a.Answered.IsZero() { + return + } + r := ix.TakeIn(ctx, a.Module) + a.Result, a.Answered = &r, time.Now().UTC() + logf("%s: a take-in asked by %s (%s): %s", a.Module, a.By, a.Why, r.Answer) + data, err := json.Marshal(a) + if err != nil { + return + } + if _, err := kv.Update(ctx, e.Key(), data, e.Revision()); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + logf("%s.%s: the answer could not be written: %v", index.TakeInsState, e.Key(), err) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index a40eeb1..592b309 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -8,6 +8,7 @@ import ( "slices" "strings" "sync" + "sync/atomic" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -65,6 +66,10 @@ type Indexer struct { previews map[string]delivery.Preview // by previews key (previews.go) toolchains map[string]map[string]delivery.Toolchain // language, then node (toolchains.go) tried map[string]bool // repository, commit and directory a registered module's manifest was tried at (registered.go) + unread map[string]bool // registered modules whose last read the forge could not answer, read again by RetryUnread + retrying atomic.Bool // a RetryUnread runs in the background (kickRetry) + 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 } @@ -89,6 +94,7 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { if err := ix.applyMerged(ctx, f); err != nil { return err } + ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945) } var digests map[string]string var held map[string]delivery.Module diff --git a/modules/mesh-delivery-indexer/internal/index/registered.go b/modules/mesh-delivery-indexer/internal/index/registered.go index 3223844..e83fb4f 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered.go +++ b/modules/mesh-delivery-indexer/internal/index/registered.go @@ -3,6 +3,8 @@ package index import ( "context" "fmt" + "maps" + "slices" "strings" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -27,20 +29,35 @@ type RepoRead struct { // IndexRegistered writes the module of a registration as a merge of its repository would, when the modules bucket holds // none of that name. Its commit is the head the gate's source names for the repository, else the registration's. The // repository's other modules are neither written nor dropped: they are the merges'. A manifest that is not at the path or -// names another module is said and left. +// names another module is said and left. A forge that could not say (a timeout, or a 404 while the forge's account could +// not read the repository) is an error, and the module is kept as unread: it is read again at the next source revision +// and the next merged message (RetryUnread), not only when the indexer starts again (issue 1945). func (ix *Indexer) IndexRegistered(ctx context.Context, r Registration) error { + _, err := ix.indexRegistered(ctx, r, false) + return err +} + +// taken is what indexRegistered did with one registration: the module written, held already, or the reason it was not. +type taken struct { + stored, held bool + repository, commit, dir string + why string +} + +func (ix *Indexer) indexRegistered(ctx context.Context, r Registration, force bool) (taken, error) { if ix.Modules == nil || ix.Read == nil { - return nil + return taken{why: "the indexer reads no forge or writes no modules"}, nil } key, err := delivery.ModulesKey(r.Module) if err != nil { ix.logf("%s: the registered module has no usable key, skipped: %v", r.Module, err) - return nil + return taken{why: fmt.Sprintf("the module has no usable key: %v", err)}, nil } if _, _, found, err := ix.Modules.Get(ctx, key); err != nil { - return fmt.Errorf("modules.%s: %w", key, err) + return taken{}, fmt.Errorf("modules.%s: %w", key, err) } else if found { - return nil + ix.markRead(r.Module) + return taken{held: true, why: "the modules state holds it already: a merge to its repository writes it from now on"}, nil } repository := delivery.RepositoryName(r.Repository) commit := r.Commit @@ -57,17 +74,20 @@ func (ix *Indexer) IndexRegistered(ctx context.Context, r Registration) error { } } ix.mu.Unlock() + t := taken{repository: repository, commit: commit, dir: dir} if repository == "" || commit == "" { ix.logf("%s: registered with no repository or commit, so its manifest is not read", r.Module) - return nil + t.why = "the controller's record names no repository or commit" + return t, nil } // One try for each repository, commit and directory: a repository with no manifest where it was registered is said once, - // not at every record of the controller's builds. + // not at every record of the controller's builds. A person's take-in (force) reads again. tried := repository + "@" + commit + ":" + dir ix.mu.Lock() - if ix.tried[tried] { + if ix.tried[tried] && !force { ix.mu.Unlock() - return nil + t.why = "tried already at this commit and directory" + return t, nil } if ix.tried == nil { ix.tried = map[string]bool{} @@ -78,17 +98,24 @@ func (ix *Indexer) IndexRegistered(ctx context.Context, r Registration) error { if err != nil { ix.mu.Lock() delete(ix.tried, tried) // a forge that could not say is asked again + if ix.unread == nil { + ix.unread = map[string]bool{} + } + ix.unread[r.Module] = true // and read again on the next source revision or merged message (RetryUnread) ix.mu.Unlock() - return fmt.Errorf("%s: reading %s at %s: %w", r.Module, repository, commit, err) + return t, fmt.Errorf("%s: reading %s at %s: %w", r.Module, repository, commit, err) } + ix.markRead(r.Module) raw, ok := read.Manifests[dir] if !ok || read.Trees[dir] == "" { ix.logf("%s: %s at %s holds no module.json at %q, so the registered module is not indexed", r.Module, repository, commit, dir) - return nil + t.why = fmt.Sprintf("%s at %s holds no module.json at %q", repository, commit, dir) + return t, nil } if name := manifestModule(raw); name != r.Module { ix.logf("%s: the manifest at %q of %s at %s names the module %q, so the registered module is not indexed", r.Module, dir, repository, commit, name) - return nil + t.why = fmt.Sprintf("the manifest at %q of %s at %s names the module %q", dir, repository, commit, name) + return t, nil } f := delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: repository, Merge: commit, Root: read.Root, Trees: map[string]string{dir: read.Trees[dir]}, @@ -96,7 +123,123 @@ func (ix *Indexer) IndexRegistered(ctx context.Context, r Registration) error { if l := read.Locks[dir]; len(l) > 0 { f.Locks = map[string]map[string]string{dir: l} } - return ix.applyManifests(ctx, f, false) + if err := ix.applyManifests(ctx, f, false); err != nil { + return t, err + } + if _, _, found, err := ix.Modules.Get(ctx, key); err != nil { + return t, fmt.Errorf("modules.%s: %w", key, err) + } else if !found { + t.why = "its manifest was read but the modules state holds no value of it (the indexer's log says why)" + return t, nil + } + t.stored = true + ix.logf("%s: the registered module is indexed from %s at %s", r.Module, repository, commit) + return t, nil +} + +// markRead takes a module out of the unread ones: the forge answered for it. +func (ix *Indexer) markRead(module string) { + ix.mu.Lock() + delete(ix.unread, module) + ix.mu.Unlock() +} + +// RetryUnread reads again every registered module whose last read the forge could not answer (a timeout, or a 404 while +// the forge's account could not read the repository; issue 1945). It is called on the events that already reach the +// indexer, a new source revision and a merged message, so a repository that became readable is taken in without a hand +// act and without a timer of its own (ADR 0437 §8). What still cannot be read is said and kept for the next. +func (ix *Indexer) RetryUnread(ctx context.Context) { + ix.mu.Lock() + names := slices.Sorted(maps.Keys(ix.unread)) + ix.mu.Unlock() + for _, module := range names { + if err := ix.IndexRegisteredModule(ctx, module); err != nil { + ix.logf("the registered module %s is still not indexed: %v", module, err) + } + } +} + +// kickRetry runs RetryUnread in one background goroutine, so the message or revision that asked it is acknowledged without +// waiting on the forge (each unread module costs up to two forge calls of 30 s each). A kick while one runs makes it run +// once more when it ends; never two at once. +func (ix *Indexer) kickRetry(ctx context.Context) { + ix.mu.Lock() + none := len(ix.unread) == 0 + ix.mu.Unlock() + if none { + return + } + ix.retryAgain.Store(true) + if !ix.retrying.CompareAndSwap(false, true) { + return + } + ix.retryWG.Add(1) + go func() { + defer ix.retryWG.Done() + for ix.retryAgain.Swap(false) { + ix.RetryUnread(ctx) + } + ix.retrying.Store(false) + // A kick between the last Swap and the Store found it running and returned: run for it. + if ix.retryAgain.Load() { + ix.kickRetry(ctx) + } + }() +} + +// TakenIn is the answer of a take-in (the tool indexer_take_in): whether the module is now in the modules state, from +// where, and otherwise why not. +type TakenIn struct { + Module string `json:"module"` + Stored bool `json:"stored"` + Held bool `json:"held,omitempty"` + Repository string `json:"repository,omitempty"` + Commit string `json:"commit,omitempty"` + Path string `json:"path,omitempty"` + Answer string `json:"answer"` +} + +// TakeIn reads a registered module's manifest now, whatever was tried before, and writes it as a merge would (the tool +// indexer_take_in, issue 1945). A module the modules state holds is answered so and left; one with no registration in the +// controller's records, or whose every registration could not be read or holds no manifest of it, is answered with why. +func (ix *Indexer) TakeIn(ctx context.Context, module string) TakenIn { + out := TakenIn{Module: module} + if ix.Builds == nil { + out.Answer = "the indexer reads no controller's build records, so it knows no registration" + return out + } + regs := ix.Builds.RegistrationsOf(module) + if len(regs) == 0 { + if key, err := delivery.ModulesKey(module); err == nil && ix.Modules != nil { + if _, _, found, err := ix.Modules.Get(ctx, key); err == nil && found { + out.Held, out.Answer = true, "the modules state holds it already: a merge to its repository writes it from now on" + return out + } + } + out.Answer = "the controller's build records name no registration of " + module + ": a module from outside the indexed repositories is known only from them, and one inside them from a merge" + return out + } + var whys []string + for _, r := range regs { + t, err := ix.indexRegistered(ctx, r, true) + out.Repository, out.Commit, out.Path = t.repository, t.commit, pathOf(t.dir) + switch { + case err != nil: + whys = append(whys, err.Error()) + continue + case t.held: + out.Held, out.Answer = true, t.why + out.Repository, out.Commit, out.Path = "", "", "" + return out + case t.stored: + out.Stored = true + out.Answer = fmt.Sprintf("written to the modules state from %s at %s", t.repository, t.commit) + return out + } + whys = append(whys, t.why) + } + out.Answer = "not taken in: " + strings.Join(whys, "; ") + return out } // IndexRegisteredModule indexes the registration the controller's records name for a module, when there is one. diff --git a/modules/mesh-delivery-indexer/internal/index/registered_test.go b/modules/mesh-delivery-indexer/internal/index/registered_test.go index 06e2838..13eca05 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered_test.go +++ b/modules/mesh-delivery-indexer/internal/index/registered_test.go @@ -3,6 +3,7 @@ package index import ( "context" "errors" + "strings" "testing" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -156,24 +157,82 @@ func TestARegistrationOfAnotherRepositoryDoesNotHideTheRealOne(t *testing.T) { / } func TestAForgeThatCannotSayLeavesTheModuleUnwrittenAndAsksAgain(t *testing.T) { // a timeout or a 404 at the head: no read at an older commit - for _, failure := range []string{"timeout", "Gitea API /repos/x/git/trees/y: 404 not found"} { - ix, mods := modulesIndexer(t, rec(t, "build-1", "snake", "http://novox.internal:20000/jschoubben/snake.git", "", "master", "", "2026-10-19T09:00:00Z", "")) - ix.Manifests = newMem() - ix.source = sourceNow{rev: 1, heads: map[string]string{"jschoubben/snake": h2}} - var asked []string - ix.Read = func(_ context.Context, _, sha, _ string) (RepoRead, error) { - asked = append(asked, sha) - return RepoRead{}, errors.New(failure) - } - if err := ix.IndexRegisteredAll(context.Background()); err == nil { - t.Errorf("%q: no error", failure) - } - if _, ok := moduleOf(t, mods, "snake"); ok || len(asked) != 1 || asked[0] != h2 { - t.Errorf("%q: written %v, asked %v", failure, ok, asked) - } - if err := ix.IndexRegisteredAll(context.Background()); err == nil || len(asked) != 2 { - t.Errorf("%q: not asked again (asked %v)", failure, asked) - } + // Issue 1945: car-hunter and snake answered 404 until the forge's account could read them, and were read again only when + // the indexer started again. Each row's trigger must now take the module in once the forge reads it. + const snake = `{"module":"snake","version":"1"}` + ctx := context.Background() + for _, tc := range []struct{ failure, trigger string }{ + {"timeout", "the indexer starts again"}, + {"Gitea API /repos/x/git/trees/y: 404 not found", "the indexer starts again"}, + {"Gitea API /repos/x/git/trees/y: 404 not found", "a new source revision"}, + {"Gitea API /repos/x/git/trees/y: 404 not found", "a merged message of another repository"}, + {"Gitea API /repos/x/git/trees/y: 404 not found", "a person's take-in"}, + {"timeout", "a person's take-in"}, + } { + t.Run(tc.failure+", then "+tc.trigger, func(t *testing.T) { + ix, mods := modulesIndexer(t, rec(t, "build-1", "snake", "http://novox.internal:20000/jschoubben/snake.git", "", "master", "", "2026-10-19T09:00:00Z", "")) + ix.Manifests = newMem() + ix.source = sourceNow{rev: 1, heads: map[string]string{"jschoubben/snake": h2}} + var asked []string + readable := false + ix.Read = func(_ context.Context, _, sha, _ string) (RepoRead, error) { + asked = append(asked, sha) + if readable { + return RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": snake}}, nil + } + return RepoRead{}, errors.New(tc.failure) + } + if err := ix.IndexRegisteredAll(ctx); err == nil { + t.Errorf("no error") + } + if _, ok := moduleOf(t, mods, "snake"); ok || len(asked) != 1 || asked[0] != h2 { + t.Errorf("written %v, asked %v", ok, asked) + } + if err := ix.IndexRegisteredAll(ctx); err == nil || len(asked) != 2 { + t.Errorf("not asked again (asked %v)", asked) + } + readable = true // the forge's account reads the repository now + var got TakenIn + switch tc.trigger { + case "the indexer starts again": + if err := ix.IndexRegisteredAll(ctx); err != nil { + t.Fatal(err) + } + case "a new source revision": + if err := ix.ApplySource(ctx, 2, delivery.Source{Heads: map[string]string{"jschoubben/snake": h2, "novox/other": h1}}); err != nil { + t.Fatal(err) + } + ix.retryWG.Wait() // read again in the background + case "a merged message of another repository": + if err := ix.Apply(ctx, mergedMsg("novox/other", map[string]string{"": "t9"})); err != nil { + t.Fatal(err) + } + ix.retryWG.Wait() // read again in the background + case "a person's take-in": + got = ix.TakeIn(ctx, "snake") + if !got.Stored || got.Repository != "jschoubben/snake" || got.Commit != h2 || got.Path != "." || got.Answer == "" { + t.Errorf("take-in answered %+v", got) + } + } + if m, ok := moduleOf(t, mods, "snake"); !ok || m.Manifest == "" || m.Commit != h2 { + t.Fatalf("after %s the readable module is not taken in: %+v, found %v (asked %v)", tc.trigger, m, ok, asked) + } + // Taken in, it is read no more: a further event asks the forge nothing. + n := len(asked) + if err := ix.ApplySource(ctx, 3, delivery.Source{Heads: map[string]string{"jschoubben/snake": h2}}); err != nil { + t.Fatal(err) + } + ix.retryWG.Wait() + if len(asked) != n { + t.Errorf("the forge was asked again after the module was taken in: %v", asked) + } + if again := ix.TakeIn(ctx, "snake"); again.Stored || !again.Held || len(asked) != n { + t.Errorf("a take-in of a held module answered %+v and asked %v", again, asked) + } + if none := ix.TakeIn(ctx, "nobody"); none.Stored || none.Held || !strings.Contains(none.Answer, "no registration") { + t.Errorf("a take-in of an unregistered module answered %+v", none) + } + }) } } diff --git a/modules/mesh-delivery-indexer/internal/index/source.go b/modules/mesh-delivery-indexer/internal/index/source.go index 38869ec..9b3e409 100644 --- a/modules/mesh-delivery-indexer/internal/index/source.go +++ b/modules/mesh-delivery-indexer/internal/index/source.go @@ -47,6 +47,10 @@ func (ix *Indexer) ApplySource(ctx context.Context, rev uint64, s delivery.Sourc if ix.Modules == nil { return ix.RecomputeCandidates(ctx) } + // A registered module the forge could not read is read again at each new source revision (issue 1945), in the + // background: a forge that does not answer must not hold this revision. One taken in later is stamped when it is written + // (applyManifests reads srcOf). + ix.kickRetry(ctx) held, err := ix.heldModules(ctx) if err != nil { return err diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go new file mode 100644 index 0000000..512bc9a --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -0,0 +1,64 @@ +package index + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The indexer's doing tool (novox/hq issue 1945; the instruction "tools are how a module is spoken to"): indexer_take_in +// reads a registered module's manifest now and writes it to the modules state. The tools bundle serves it through the tool +// runner; the indexer's process serves no request subject (ADR 0437 §4), so the tool writes an ask into the indexer's +// state takeins, which the indexer watches and answers there, as the builder answers builder_ask_again. The read tools +// of the indexer's states stay in mesh-delivery-view. +const ( + TakeInTool = "indexer_take_in" + // TakeInsState is the indexer's state of take-in asks: the bucket mesh-delivery-indexer_takeins, keyed as the modules state. + TakeInsState = "takeins" + TakeInsBucket = delivery.IndexerModule + "_" + TakeInsState +) + +// TakeInAsk is a value of the takeins state: which module a person asked to be taken in, why, who and when, and, once the +// indexer read it, its answer. +type TakeInAsk struct { + V int `json:"v"` + Module string `json:"module"` + Why string `json:"why,omitempty"` + By string `json:"by,omitempty"` + At time.Time `json:"at"` + Result *TakenIn `json:"result,omitempty"` + Answered time.Time `json:"answered,omitzero"` +} + +// DecodeTakeIn reads a takeins value. +func DecodeTakeIn(data []byte) (TakeInAsk, error) { + var a TakeInAsk + if err := json.Unmarshal(data, &a); err != nil { + return a, err + } + if a.V != 1 { + return a, fmt.Errorf("a takeins value of version %d, not 1", a.V) + } + if a.Module == "" { + return a, errors.New("a takeins value naming no module") + } + return a, nil +} + +// NewTakeIn is the value indexer_take_in writes for a module: its key and the value. +func NewTakeIn(module, why, by string, at time.Time) (string, []byte, error) { + module = strings.TrimSpace(module) + if module == "" { + return "", nil, errors.New("name the module to take in (module)") + } + key, err := delivery.ModulesKey(module) + if err != nil { + return "", nil, err + } + data, err := json.Marshal(TakeInAsk{V: 1, Module: module, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) + return key, data, err +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 5822f79..46b3589 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -24,18 +24,20 @@ type state struct { } type artifact struct { - Name string `json:"name"` - Kind string `json:"kind"` - Language string `json:"language"` - System string `json:"system"` - From string `json:"from"` - Binary string `json:"binary"` + Name string `json:"name"` + Kind string `json:"kind"` + Language string `json:"language"` + System string `json:"system"` + From string `json:"from"` + Binary string `json:"binary"` + Loads []string `json:"loads"` } type manifest struct { Module string `json:"module"` Version string `json:"version"` Slug string `json:"slug"` + Tools []string `json:"tools"` Consumes []string `json:"consumes"` State []state `json:"state"` Reads []string `json:"reads"` @@ -78,7 +80,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if m.Module != "mesh-delivery-indexer" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}}; !reflect.DeepEqual(m.State, want) { + if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { @@ -101,8 +103,12 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } - // No emits, no streams, no bus, no seat, no tool (ADR 0437 §4). - for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats", "bus"} { + // 1945: one tool, indexer_take_in, served by its tools bundle through the takeins state; the process serves none. + if want := []string{"indexer_take_in"}; !reflect.DeepEqual(m.Tools, want) { + t.Errorf("tools = %v, want %v", m.Tools, want) + } + // No emits, no streams, no bus, no seat (ADR 0437 §4). + for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the indexer has none", k) } @@ -134,8 +140,13 @@ func TestManifestProcess(t *testing.T) { // 1238/AC1 if _, bus := p["bus"]; bus { t.Errorf("the process declares a bus: it runs on its own broker credential") } - if len(m.Build.Artifacts) != 1 { - t.Fatalf("artifacts = %+v", m.Build.Artifacts) + if len(m.Build.Artifacts) != 2 { + t.Fatalf("artifacts = %+v, want the program and the tools", m.Build.Artifacts) + } + if a := m.Build.Artifacts[1]; a.Name != "tools" || a.Kind != "bundle" || a.Binary != "mesh-delivery-indexer-tools" || !reflect.DeepEqual(a.Loads, []string{"mesh-delivery-indexer-tools"}) { + t.Errorf("tools artifact = %+v", a) // 1945 + } else if st, err := os.Stat(a.From); err != nil || !st.IsDir() { + t.Errorf("the tools' source %s: %v", a.From, err) } a := m.Build.Artifacts[0] if st, err := os.Stat(a.From); err != nil || !st.IsDir() || a.Binary != "mesh-delivery-indexer" || !strings.HasSuffix(a.From, a.Binary) { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 6abdf0e..a30c24e 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -2,6 +2,9 @@ "module": "mesh-delivery-indexer", "version": "1", "slug": "indexer", + "tools": [ + "indexer_take_in" + ], "consumes": [ "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", @@ -11,7 +14,8 @@ "state": [ { "name": "candidates", "history": 10 }, { "name": "manifests", "history": 1 }, - { "name": "modules", "history": 20 } + { "name": "modules", "history": 20 }, + { "name": "takeins", "history": 1 } ], "reads": [ "mesh-build-go.toolchains", @@ -72,6 +76,17 @@ "system": "arch", "from": "cmd/mesh-delivery-indexer", "binary": "mesh-delivery-indexer" + }, + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-delivery-indexer-tools", + "binary": "mesh-delivery-indexer-tools", + "loads": [ + "mesh-delivery-indexer-tools" + ] } ] }