mesh-delivery-indexer reads a registered module again once its repository becomes readable, and takes one in on ask (issue 1945)

A registered module whose repository the forge answered 404 (car-hunter, snake, before the forge's account could read
them) was read again only when the indexer started again or the controller wrote a new build record of it, so the grants
part's composed user list stayed short of their subjects and the take-over was refused. A failed read now keeps the
module as unread, read again on the next source revision and the next merged message (no timer of its own), and the new
tool indexer_take_in (the tools bundle mesh-delivery-indexer-tools, through the indexer's takeins state, as the
builder's builder_ask_again) takes one in now and answers what it stored or why not.
This commit is contained in:
2026-10-26 14:01:34 +01:00
parent 1d41682715
commit 07b5f13474
10 changed files with 477 additions and 44 deletions
@@ -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 %s.%s within %s: it is not running, or watches no %s (it says so in its journal at start)", index.TakeInsState, key, bound, index.TakeInsBucket)
}
}
}
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)
}
}
@@ -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.<group>): 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)
@@ -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)
}
}
@@ -65,6 +65,7 @@ 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
recompute sync.Mutex // one Recompute at a time
}
@@ -89,6 +90,7 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error {
if err := ix.applyMerged(ctx, f); err != nil {
return err
}
ix.RetryUnread(ctx) // a registered module the forge could not read is read again on a merge (issue 1945)
}
var digests map[string]string
var held map[string]delivery.Module
@@ -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,95 @@ 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)
}
}
}
// 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.
@@ -3,6 +3,7 @@ package index
import (
"context"
"errors"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -156,24 +157,79 @@ 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)
}
case "a merged message of another repository":
if err := ix.Apply(ctx, mergedMsg("novox/other", map[string]string{"": "t9"})); err != nil {
t.Fatal(err)
}
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)
}
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)
}
})
}
}
@@ -47,6 +47,9 @@ 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), before the
// stamping below, so one taken in at the head is stamped too.
ix.RetryUnread(ctx)
held, err := ix.heldModules(ctx)
if err != nil {
return err
@@ -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
}
+22 -11
View File
@@ -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) {
+16 -1
View File
@@ -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"
]
}
]
}