The indexer fixes its own build once at its start and never learns it later; a merge held for an unknown own build is said pending with why and since (issue 1921, ADR 0633 §1)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -216,7 +216,7 @@ var errReadOnly = errors.New("the tools bundle only reads this state")
|
||||
func main() {
|
||||
ask := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second)
|
||||
tools := append(Tools(ask, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...)
|
||||
tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, ask, time.Now)...)
|
||||
tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, keptRead{stdio.State(index.OwnBuildStateName)}, ask, time.Now)...)
|
||||
tools = append(tools, index.TryCandidateTools(keptRead{stdio.State("candidates")}, keptRead{stdio.State(index.TryRecoveryState)}, ask, time.Now)...)
|
||||
if err := stdio.Serve("", tools); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err)
|
||||
|
||||
@@ -283,7 +283,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
|
||||
}
|
||||
waits := func() []takein.PendingWait {
|
||||
t.Helper()
|
||||
answer, err := index.PendingTools(kvStore{pendingKV}, nil, time.Now)[0].Run(map[string]any{})
|
||||
answer, err := index.PendingTools(kvStore{pendingKV}, nil, nil, time.Now)[0].Run(map[string]any{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -343,8 +343,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
oldSections := process.AwaitBucket(ctx, js, conn, index.OldSectionsBucket, "the count of the old sections is not kept, so the judge refuses no use of one", logger.Printf)
|
||||
// The module graph (ADR 0603 §2): a state this delivery brings, bound when it appears; until then no graph is written.
|
||||
graph := process.AwaitBucket(ctx, js, conn, graph.Bucket, "no module graph is written", logger.Printf)
|
||||
ownBuildState := process.AwaitBucket(ctx, js, conn, index.OwnBuildBucket, "which build this indexer runs is not kept", logger.Printf)
|
||||
tryRecovery := process.AwaitBucket(ctx, js, conn, index.TryRecoveryBucket, "how the recovery of the open tries went is not kept", logger.Printf)
|
||||
ix := &index.Indexer{TryRecovery: awaitedStore{tryRecovery}, OldSections: awaitedStore{oldSections}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known,
|
||||
ix := &index.Indexer{OwnBuildState: awaitedStore{ownBuildState}, TryRecovery: awaitedStore{tryRecovery}, OldSections: awaitedStore{oldSections}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known,
|
||||
Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(js, cred, logger.Printf), SayAccepted: sayAccepted(js, cred, logger.Printf),
|
||||
Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf,
|
||||
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
|
||||
@@ -354,6 +355,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// until then every merge is taken in at once, as before.
|
||||
Pending: awaitedStore{process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)},
|
||||
SayTakeIn: sayTakeIn(js, cred)}
|
||||
// Which build this program is, fixed here once, before it hears any merge and never asked again (index.FixOwnBuild): the
|
||||
// node-engine names the successor while it replaces this process, and an old process that learnt it later would take
|
||||
// revisions in with its old judge.
|
||||
ix.FixOwnBuild(ctx)
|
||||
// The log's entries at the start are read before any fingerprint is computed or registered module indexed (below); each
|
||||
// new one indexes the module it names when that module is registered and the bucket holds none (registered.go).
|
||||
var started atomic.Bool
|
||||
|
||||
@@ -122,12 +122,17 @@ type Indexer struct {
|
||||
fillAgain atomic.Bool // a kick arrived while it ran
|
||||
atHeads map[string]map[string]delivery.PublishedPackages // a member head's published Go modules, by repository@head (candidatePublished)
|
||||
// OwnBuild answers the fingerprint of the build this program runs, from the node-engine's seat; nil knows none.
|
||||
OwnBuild func(ctx context.Context) (string, error)
|
||||
underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go)
|
||||
judgeMu sync.Mutex // one judge pass at a time (verdicts.go)
|
||||
ownMu sync.Mutex // guards ownFP
|
||||
ownFP string // the fingerprint of the build this program runs, asked at the start pass (learnOwnBuild)
|
||||
judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go)
|
||||
OwnBuild func(ctx context.Context) (string, error)
|
||||
underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go)
|
||||
judgeMu sync.Mutex // one judge pass at a time (verdicts.go)
|
||||
ownMu sync.Mutex // guards ownFP
|
||||
ownFP string // the fingerprint of the build this program runs, fixed once at the start (FixOwnBuild)
|
||||
ownFixed bool // FixOwnBuild asked already: it asks once
|
||||
ownSince time.Time // when the build was asked and not known, with ownWhy
|
||||
ownWhy string
|
||||
// OwnBuildState keeps which build this process knows it runs, or since when and why not (FixOwnBuild); nil keeps none.
|
||||
OwnBuildState Store
|
||||
judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go)
|
||||
// Message asks the forge bridge for the merged message of a merge, repository and merge commit in, the message whole out (a
|
||||
// pending value keeps a reference, not a copy: issue 1585); an error is the bridge's words. Nil asks none.
|
||||
Message func(ctx context.Context, repository, merge string) ([]byte, error)
|
||||
|
||||
@@ -339,20 +339,27 @@ func (ix *Indexer) pendingOnBus(ctx context.Context) bool {
|
||||
return !errors.Is(err, process.ErrNotYet)
|
||||
}
|
||||
|
||||
// ownBuildNow is the build this program runs: the one learnt, else asked of the node-engine now.
|
||||
func (ix *Indexer) ownBuildNow(ctx context.Context) string {
|
||||
ix.ownMu.Lock()
|
||||
fp := ix.ownFP
|
||||
ix.ownMu.Unlock()
|
||||
if fp != "" || ix.OwnBuild == nil {
|
||||
return fp
|
||||
}
|
||||
ix.learnOwnBuild(ctx)
|
||||
// ownBuildNow is the build this program runs, as FixOwnBuild fixed it at the start: empty while unknown, and never asked again.
|
||||
func (ix *Indexer) ownBuildNow(context.Context) string {
|
||||
ix.ownMu.Lock()
|
||||
defer ix.ownMu.Unlock()
|
||||
return ix.ownFP
|
||||
}
|
||||
|
||||
// sayOwnBuildUnknown says, for a merge this process cannot take in because it does not know its own build (ADR 0633 §1), a
|
||||
// take-in.pending that names why and since when, so the watchman's take-in-waits bound covers it. The merge is not written to
|
||||
// pending; the stream delivers it again to the process that knows its build.
|
||||
func (ix *Indexer) sayOwnBuildUnknown(ctx context.Context, seq uint64, f forgemsg.Forge) {
|
||||
ix.ownMu.Lock()
|
||||
since, why := ix.ownSince, ix.ownWhy
|
||||
ix.ownMu.Unlock()
|
||||
if since.IsZero() { // never fixed: the start did not run (no engine wired)
|
||||
since, why = ix.now(), "the build this program runs was never asked"
|
||||
}
|
||||
ix.sayTakeIn(ctx, takein.TakeIn{Kind: takein.TakeInPending, Key: takein.PendingKey(seq), What: takein.TakeInRevision,
|
||||
Merges: []takein.Merge{{Repository: f.Repository, Number: f.Number, Merge: f.Merge}}, Since: since, Unknown: why})
|
||||
}
|
||||
|
||||
// sayTakeIn publishes a take-in event and keeps what it said on the entries, so indexer_pending reads the last event of
|
||||
// each wait from the state. A publish that fails is said in the log: the next evaluation says it again.
|
||||
func (ix *Indexer) sayTakeIn(ctx context.Context, t takein.TakeIn) bool {
|
||||
@@ -401,6 +408,7 @@ func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f forgemsg.Forge) (Ap
|
||||
}
|
||||
own := ix.ownBuildNow(ctx)
|
||||
if own == "" {
|
||||
ix.sayOwnBuildUnknown(ctx, seq, f)
|
||||
return Applied{}, "", ErrOwnBuildUnknown
|
||||
}
|
||||
ix.pendingMu.Lock()
|
||||
@@ -691,7 +699,7 @@ func (ix *Indexer) dry() *dryRun {
|
||||
d := &dryRun{}
|
||||
x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum,
|
||||
GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains,
|
||||
ReadOpen: nil, OwnBuild: nil, Message: ix.Message, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil, ReadFile: ix.ReadFile, underSeats: true}
|
||||
ReadOpen: nil, OwnBuild: nil, OwnBuildState: nil, Message: ix.Message, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil, ReadFile: ix.ReadFile, underSeats: true}
|
||||
x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) }
|
||||
x.Store, d.candidates = over(ix.Store)
|
||||
x.Modules, d.modules = over(ix.Modules)
|
||||
|
||||
@@ -61,6 +61,8 @@ func newTakeInBed(t *testing.T) *takeInBed {
|
||||
t.Fatalf("the bed's modules have no fingerprint")
|
||||
}
|
||||
ix.OwnBuild = func(context.Context) (string, error) { return own, nil }
|
||||
ix.OwnBuildState = newMem()
|
||||
ix.FixOwnBuild(context.Background()) // fixed once, at the start
|
||||
return b
|
||||
}
|
||||
|
||||
@@ -131,11 +133,26 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
ctx := context.Background()
|
||||
t.Run("T6 a merge while the own build is unknown writes nothing", func(t *testing.T) {
|
||||
b := newTakeInBed(t)
|
||||
b.ix.ownFP, b.ix.ownFixed = "", false // this process is started again and asks the engine, which does not answer
|
||||
b.ix.OwnBuild = func(context.Context) (string, error) { return "", errors.New("the engine does not answer") }
|
||||
b.ix.FixOwnBuild(ctx)
|
||||
b.said = nil
|
||||
_, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", ""))
|
||||
if !errors.Is(err, ErrOwnBuildUnknown) || len(b.keys()) != 0 || b.commit(t, "avahi") == sha('b') {
|
||||
t.Errorf("err %v, pending %v, avahi at %s: want ErrOwnBuildUnknown and nothing written", err, b.keys(), b.commit(t, "avahi"))
|
||||
}
|
||||
// The merge is said pending with why and since when (so the watchman's bound covers it), and indexer_pending says the
|
||||
// build is unknown, from the own-build state.
|
||||
if len(b.said) != 1 || b.said[0].Kind != takein.TakeInPending || b.said[0].Key != takein.PendingKey(1) || !strings.Contains(b.said[0].Unknown, "does not answer") ||
|
||||
b.said[0].Since.IsZero() || b.said[0].WaitsFor != "" {
|
||||
t.Errorf("said %+v, want one take-in.pending of m.1 saying why the own build is unknown", b.said)
|
||||
} else if _, err := takein.EncodeTakeIn(b.said[0]); err != nil {
|
||||
t.Errorf("the unknown-build event is refused by its own contract: %v", err)
|
||||
}
|
||||
answer, err := PendingTools(b.pending, b.ix.OwnBuildState, nil, nil)[0].Run(map[string]any{})
|
||||
if a, _ := answer.(takein.PendingAnswer); err != nil || !strings.Contains(a.OwnBuildUnknown, "does not answer") || a.OwnBuildSince.IsZero() {
|
||||
t.Errorf("indexer_pending = %+v %v, want the build unknown with why and since", answer, err)
|
||||
}
|
||||
// Before the pending state is on the bus nothing waits: the start asks no build, and the merge is taken in at once,
|
||||
// as before ADR 0633, though the build is unknown (TestIndexerWhole found the start waiting on it for ever).
|
||||
b.ix.Pending = &memStore{unbound: process.ErrNotYet}
|
||||
@@ -254,31 +271,36 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
g.Members = []delivery.Head{{Repository: "novox/mesh-sdk", Number: 30, Head: g.Head}, {Repository: "novox/mesh-host", Number: 4, Head: sha('9')}}
|
||||
b.arrive(t, 5, g)
|
||||
named := b.fp(t, delivery.IndexerModule)
|
||||
// The new indexer starts; its build is not known at first: a hand ask is answered meanwhile (R5), and the start waits.
|
||||
next := &Indexer{}
|
||||
*next = Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
|
||||
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: b.ix.Message, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum}
|
||||
next.toolchains = b.ix.toolchains
|
||||
asked := 0
|
||||
next.OwnBuild = func(context.Context) (string, error) {
|
||||
if asked++; asked < 3 { // the hand ask's and the first start's
|
||||
return "", errors.New("not yet")
|
||||
}
|
||||
return named, nil
|
||||
// The new indexer starts; its build is not known (the engine does not answer at its start): a hand ask is answered
|
||||
// meanwhile (R5), and the start waits for nothing. It is never asked again, so it stays unknown until it is started again.
|
||||
mk := func(ask func(context.Context) (string, error)) *Indexer {
|
||||
n := &Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
|
||||
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: b.ix.Message, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum}
|
||||
n.toolchains = b.ix.toolchains
|
||||
n.OwnBuild = ask
|
||||
n.FixOwnBuild(ctx)
|
||||
return n
|
||||
}
|
||||
asked := 0
|
||||
next := mk(func(context.Context) (string, error) { asked++; return "", errors.New("not yet") })
|
||||
if r := next.TakeInPendingByHand(ctx, "m.nothing", "lead@shanks", "a test"); r.Stored || !strings.Contains(r.Answer, "nothing waits") {
|
||||
t.Errorf("a hand ask while the own build is unknown: %+v", r)
|
||||
}
|
||||
// The first start, its build not known, waits for nothing (no sleep, no timer): it answers at once, takes nothing
|
||||
// in, and the forge consumer starts after it.
|
||||
if _, words, err := next.TakeInAtStart(ctx); err != nil || asked != 2 || len(b.keys()) != 2 || !strings.Contains(words, "not known") {
|
||||
t.Fatalf("a start with its build unknown: %q %v, asked %d, pending %v: want it to answer at once and take nothing in", words, err, asked, b.keys())
|
||||
if _, words, err := next.TakeInAtStart(ctx); err != nil || len(b.keys()) != 2 || !strings.Contains(words, "not known") {
|
||||
t.Fatalf("a start with its build unknown: %q %v, pending %v: want it to answer at once and take nothing in", words, err, b.keys())
|
||||
}
|
||||
// The engine answers now, and the same process does not ask it again: it learnt nothing it did not know at its start.
|
||||
next.OwnBuild = func(context.Context) (string, error) { asked++; return named, nil }
|
||||
if _, words, err := next.TakeInAtStart(ctx); err != nil || asked != 1 || len(b.keys()) != 2 || !strings.Contains(words, "not known") {
|
||||
t.Fatalf("the window of a replacement: %q %v, asked %d, pending %v: want the build fixed at the start, never learnt later", words, err, asked, b.keys())
|
||||
}
|
||||
// The module started again fixes the build at its start and takes the revision in.
|
||||
next = mk(func(context.Context) (string, error) { return named, nil })
|
||||
if _, _, err := next.TakeInAtStart(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if asked < 3 || !reflect.DeepEqual(b.keys(), []string{takein.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') {
|
||||
t.Errorf("start: asked %d, pending %v, avahi at %s: want the revision taken in, the group left", asked, b.keys(), b.commit(t, "avahi"))
|
||||
if !reflect.DeepEqual(b.keys(), []string{takein.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') {
|
||||
t.Errorf("start: pending %v, avahi at %s: want the revision taken in, the group left", b.keys(), b.commit(t, "avahi"))
|
||||
}
|
||||
})
|
||||
t.Run("T8 T9 T11 a group gathered until its end, a merge behind it, one alone, an end again", func(t *testing.T) {
|
||||
@@ -452,7 +474,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
|
||||
b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2"))
|
||||
var answer takein.PendingAnswer
|
||||
for _, tool := range PendingTools(b.pending, nil, nil) {
|
||||
for _, tool := range PendingTools(b.pending, nil, nil, nil) {
|
||||
if tool.Name == PendingTool {
|
||||
out, err := tool.Run(map[string]any{})
|
||||
if err != nil {
|
||||
@@ -522,6 +544,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: bridgeDown, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf,
|
||||
GoSum: b.ix.GoSum, OwnBuild: func(context.Context) (string, error) { return named, nil }}
|
||||
next.toolchains = b.ix.toolchains
|
||||
next.FixOwnBuild(ctx)
|
||||
b.said = nil
|
||||
if _, _, err := next.TakeInAtStart(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -578,7 +601,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
named := map[string]bool{}
|
||||
for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections",
|
||||
"JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk",
|
||||
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", "ReadFile", "OpenTries", "TryRecovery"} {
|
||||
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", "ReadFile", "OpenTries", "TryRecovery", "OwnBuildState"} {
|
||||
named[f] = true
|
||||
}
|
||||
ty := reflect.TypeOf(Indexer{})
|
||||
|
||||
@@ -16,7 +16,7 @@ import (
|
||||
// PendingTools are the indexer's tools of what waits for a take-in (novox/hq ADR 0633 §5): indexer_pending reads the
|
||||
// pending state, and indexer_take_in_pending and indexer_drop_pending steer it by an ask in takeins, as indexer_take_in
|
||||
// does. pending is the pending state as the tools bundle reads it; ask writes an ask and waits for the indexer's answer.
|
||||
func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool {
|
||||
func PendingTools(pending, ownBuild Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool {
|
||||
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s }
|
||||
hand := func(drop bool) func(map[string]any) (any, error) {
|
||||
return func(args map[string]any) (any, error) {
|
||||
@@ -53,7 +53,8 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk,
|
||||
"merges, since when it waits, the modules it holds back (holds), the merges whose message the forge bridge's stream no " +
|
||||
"longer holds (lost: they are not taken in; indexer_drop_pending drops them) and the last take-in event said of it. Empty waits: " +
|
||||
"every merge is taken in. The watchman raises take-in-waits past 60 minutes for a revision, 30 for a group; " +
|
||||
"indexer_take_in_pending and indexer_drop_pending steer a wait.",
|
||||
"indexer_take_in_pending and indexer_drop_pending steer a wait. own-build-unknown, when set, says the indexer cannot learn the build it " +
|
||||
"runs (since when): it fixes it once at its start, so every merge waits (a take-in.pending that says so) until the module is started again.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{}},
|
||||
Run: func(map[string]any) (any, error) {
|
||||
var unread []string
|
||||
@@ -63,7 +64,17 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk,
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return takein.PendingAnswer{Waits: PendingWaits(entries), Unread: unread}, nil
|
||||
answer := takein.PendingAnswer{Waits: PendingWaits(entries), Unread: unread}
|
||||
// Whether this indexer knows the build it runs: when it does not, every merge waits, and this says since when and why.
|
||||
if ownBuild != nil {
|
||||
if v, _, found, err := ownBuild.Get(context.Background(), OwnBuildKey); err == nil && found {
|
||||
var o OwnBuild
|
||||
if json.Unmarshal(v, &o) == nil && !o.Known {
|
||||
answer.OwnBuildUnknown, answer.OwnBuildSince = o.Why, o.Since
|
||||
}
|
||||
}
|
||||
}
|
||||
return answer, nil
|
||||
}},
|
||||
{Name: TakeInPendingTool,
|
||||
Description: "Take a wait in by hand with the running indexer and its judge (ADR 0633 §5): the revision under its m. key, " +
|
||||
|
||||
@@ -211,7 +211,6 @@ func (ix *Indexer) judgeOverHeld(ctx context.Context, fresh map[string]bool, ove
|
||||
// digest of the verdicts state that another judge refused or held waiting (one of a candidate, or one back on the trunk
|
||||
// later) is judged again by this judge (judgeStale), so a refusal an older judge made never stands for good.
|
||||
func (ix *Indexer) JudgeAtStart(ctx context.Context) (Judged, error) {
|
||||
ix.learnOwnBuild(ctx)
|
||||
j, err := ix.JudgeAll(ctx, nil)
|
||||
ix.judgePending.Store(err != nil)
|
||||
if err == nil {
|
||||
@@ -355,21 +354,66 @@ func (ix *Indexer) writeVerdict(ctx context.Context, digest string, v delivery.M
|
||||
return wrote, err
|
||||
}
|
||||
|
||||
// learnOwnBuild asks where the indexer's program is told which build it runs (OwnBuild: the node-engine's seat, never the
|
||||
// trunk's value of the indexer, which names a build that may be merged and not delivered). Asked at the start pass; where
|
||||
// it is not answered the build stays unknown and every unknown field waits, which is the safe side.
|
||||
func (ix *Indexer) learnOwnBuild(ctx context.Context) {
|
||||
if ix.OwnBuild == nil {
|
||||
// OwnBuildState is the indexer's state of the build it runs (the bucket mesh-delivery-indexer_own-build, one key, OwnBuildKey):
|
||||
// written once, at the process's start by FixOwnBuild, so indexer_pending says whether this process knows its build and, when it
|
||||
// does not, since when and why.
|
||||
const (
|
||||
OwnBuildStateName = "own-build"
|
||||
OwnBuildBucket = delivery.IndexerModule + "_" + OwnBuildStateName
|
||||
OwnBuildKey = "state"
|
||||
)
|
||||
|
||||
// OwnBuild is the value of the own-build state.
|
||||
type OwnBuild struct {
|
||||
V int `json:"v"`
|
||||
Known bool `json:"known"`
|
||||
FP string `json:"fp,omitempty"`
|
||||
Since time.Time `json:"since"`
|
||||
Why string `json:"why,omitempty"`
|
||||
}
|
||||
|
||||
// FixOwnBuild fixes, once and at the process's start, which build of the indexer this program is: asked of the node-engine's
|
||||
// seat (OwnBuild; never the trunk's value of the indexer, which names a build that may be merged and not delivered), before
|
||||
// this process hears a merge. It is never asked again: while the node-engine replaces the indexer the seat already names the
|
||||
// successor, and an old process that learnt it later would take revisions in with its old judge. Where the start's ask is not
|
||||
// answered the build stays unknown for the life of this process: every merge waits, loudly (a take-in.pending that says so, the
|
||||
// own-build state, indexer_pending), and the module is started again to learn it.
|
||||
func (ix *Indexer) FixOwnBuild(ctx context.Context) {
|
||||
ix.ownMu.Lock()
|
||||
if ix.ownFixed || ix.OwnBuild == nil {
|
||||
ix.ownMu.Unlock()
|
||||
return
|
||||
}
|
||||
ix.ownFixed = true
|
||||
ix.ownMu.Unlock()
|
||||
fp, err := ix.OwnBuild(ctx)
|
||||
if err != nil || fp == "" {
|
||||
ix.logf("judge: the build this program runs is not known (%v), so a manifest with a field it does not know waits and is not refused", err)
|
||||
return
|
||||
if err == nil && fp == "" {
|
||||
err = errors.New("the node-engine names no build of the indexer")
|
||||
}
|
||||
st := OwnBuild{V: 1, Since: ix.now().UTC()}
|
||||
if err != nil {
|
||||
st.Why = err.Error()
|
||||
ix.logf("judge: the build this program runs is not known (%v), so every merge waits and a manifest with a field it does not know waits and is not refused; it is learnt only at the start, so the indexer is started again", err)
|
||||
} else {
|
||||
st.Known, st.FP = true, fp
|
||||
}
|
||||
ix.ownMu.Lock()
|
||||
ix.ownFP = fp
|
||||
ix.ownFP, ix.ownSince, ix.ownWhy = fp, st.Since, st.Why
|
||||
ix.ownMu.Unlock()
|
||||
if ix.OwnBuildState == nil {
|
||||
return
|
||||
}
|
||||
data, _ := json.Marshal(st)
|
||||
if _, rev, found, gerr := ix.OwnBuildState.Get(ctx, OwnBuildKey); gerr == nil && found {
|
||||
err = ix.OwnBuildState.Update(ctx, OwnBuildKey, data, rev)
|
||||
} else if gerr == nil {
|
||||
err = ix.OwnBuildState.Create(ctx, OwnBuildKey, data)
|
||||
} else {
|
||||
err = gerr
|
||||
}
|
||||
if err != nil {
|
||||
ix.logf("the own-build state cannot be written: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ownBuildIsNewest says whether the build of the indexer this program runs (ownFP, from the node-engine) is the newest the
|
||||
@@ -510,8 +554,6 @@ func (ix *Indexer) JudgeAgain(ctx context.Context, module, digest string) TakenI
|
||||
out.Answer = "the modules state cannot be read: " + err.Error()
|
||||
return out
|
||||
}
|
||||
// The build this program runs is asked again, so a start pass that missed the engine's answer is recovered without a restart.
|
||||
ix.learnOwnBuild(ctx)
|
||||
if digest != "" {
|
||||
// One digest (a candidate's, or one the trunk no longer names), judged by this judge in the catalogue the trunk holds.
|
||||
value, _, found, err := ix.Manifests.Get(ctx, digest)
|
||||
|
||||
@@ -273,12 +273,16 @@ func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) {
|
||||
if rows, _ := ListVerdicts(keys, get, "", true); len(rows) <= len(keys)-3 || rows[0].Module > rows[len(rows)-1].Module {
|
||||
t.Fatalf("indexer_verdicts, all: %d rows of %d verdicts", len(rows), len(keys))
|
||||
}
|
||||
// indexer_judge asks the engine for the running build again: a start pass that missed it is recovered (the trunk's value of
|
||||
// the indexer is the newer one, so the field waits; asked and answered with that same build it is refused).
|
||||
ix.ownFP = ""
|
||||
// The build is fixed once, at the start, and never learnt later: this program started when the engine could not answer, and
|
||||
// the engine now names the trunk's newest build (as it does while it replaces the indexer by a successor). indexer_judge
|
||||
// does not take it for this program's own, so the field still waits and the build stays unknown.
|
||||
ix.ownFP, ix.ownFixed = "", true
|
||||
ix.OwnBuild = func(context.Context) (string, error) { return "src2:" + strings.Repeat("a", 64), nil }
|
||||
if got := ix.JudgeAgain(ctx, "later", ""); !got.Stored || !strings.Contains(got.Answer, "refused") || ix.ownFP == "" {
|
||||
t.Fatalf("indexer_judge did not learn the running build: %+v", got)
|
||||
if got := ix.JudgeAgain(ctx, "later", ""); !got.Stored || !strings.Contains(got.Answer, "waits") || ix.ownFP != "" {
|
||||
t.Fatalf("indexer_judge learnt the build the engine names now: %+v (own %q)", got, ix.ownFP)
|
||||
}
|
||||
if _, err := ix.JudgeAtStart(ctx); err != nil || ix.ownFP != "" || !verdict(later).Waits() {
|
||||
t.Fatalf("a judge pass learnt the build: %v %q %+v", err, ix.ownFP, verdict(later))
|
||||
}
|
||||
ix.ownFP = ownFP
|
||||
ix.OwnBuild = nil
|
||||
|
||||
@@ -108,7 +108,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
t.Errorf("emits = %v, want manifest-refused, manifest-accepted and the two take-in events", m.Emits)
|
||||
}
|
||||
// 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard.
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory},
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: "own-build", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory},
|
||||
{Name: graph.State, History: graph.History}, {Name: index.PendingState, History: 1}}; !reflect.DeepEqual(m.State, want) {
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
|
||||
@@ -47,6 +47,7 @@
|
||||
{ "name": "verdicts", "history": 5 },
|
||||
{ "name": "oldsections", "history": 1 },
|
||||
{ "name": "try-recovery", "history": 1 },
|
||||
{ "name": "own-build", "history": 1 },
|
||||
{ "name": "packages", "history": 5 },
|
||||
{ "name": "graph", "history": 5 },
|
||||
{ "name": "pending", "history": 1 }
|
||||
|
||||
@@ -140,6 +140,9 @@ type TakeIn struct {
|
||||
// the seat definitions the revision carries (ADR 0633 §1), the manifest validator's problems. The revision waits for
|
||||
// an indexer whose manifest is accepted, and says why here, as it says what it lost.
|
||||
Refused string `json:"refused,omitempty"`
|
||||
// Unknown is why the indexer holds a merge it does not take in although no indexer build is named: it cannot learn the build
|
||||
// it runs itself (ADR 0633 §1), so Since is when it began not to know. A pending revision names WaitsFor or Unknown.
|
||||
Unknown string `json:"own-build-unknown,omitempty"`
|
||||
}
|
||||
|
||||
func (t TakeIn) check() error {
|
||||
@@ -168,8 +171,8 @@ func (t TakeIn) check() error {
|
||||
if !strings.HasPrefix(t.Key, "m.") {
|
||||
return fmt.Errorf("delivery: a revision's take-in key is its oldest merge's m.<sequence>, not %q", t.Key)
|
||||
}
|
||||
if t.Kind == TakeInPending && t.WaitsFor == "" {
|
||||
return fmt.Errorf("delivery: a pending revision %s names the indexer build it waits for", t.Key)
|
||||
if t.Kind == TakeInPending && t.WaitsFor == "" && t.Unknown == "" {
|
||||
return fmt.Errorf("delivery: a pending revision %s names the indexer build it waits for, or why the indexer does not know its own", t.Key)
|
||||
}
|
||||
case TakeInGroup:
|
||||
if t.Group == "" {
|
||||
@@ -247,6 +250,10 @@ type PendingWait struct {
|
||||
type PendingAnswer struct {
|
||||
Waits []PendingWait `json:"waits"`
|
||||
Unread []string `json:"unread,omitempty"`
|
||||
// OwnBuildUnknown is why the running indexer does not know its own build, and since when: while it is set, every merge waits
|
||||
// (a take-in.pending with the same words) and the indexer is started again to learn it. Empty when the build is known.
|
||||
OwnBuildUnknown string `json:"own-build-unknown,omitempty"`
|
||||
OwnBuildSince time.Time `json:"own-build-since,omitzero"`
|
||||
}
|
||||
|
||||
// DecodePendingAnswer reads indexer_pending's answer, refusing a wait that is not a revision or a group, or names no
|
||||
|
||||
@@ -20,6 +20,7 @@ func sampleTakeIns(since time.Time) []TakeIn {
|
||||
Members: []delivery.Head{{Repository: "novox/mesh-delivery", Number: 9, Head: "ffff"}, {Repository: "novox/mesh-sdk", Number: 37, Head: "dddd"}}, Since: since},
|
||||
{Kind: TakeInTaken, Key: PendingKey(12), What: TakeInRevision, Merges: m, WaitsFor: "src2:abc", Since: since,
|
||||
By: "lead-47@shanks", Why: "the named indexer can never run"},
|
||||
{Kind: TakeInPending, Key: PendingKey(13), What: TakeInRevision, Merges: m, Unknown: "the engine does not answer", Since: since}, // the indexer cannot learn its own build
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -106,6 +106,24 @@ func TestTheWatchmansFindings(t *testing.T) {
|
||||
}
|
||||
expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound-10*time.Minute)
|
||||
}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}},
|
||||
{"ADR 0633 §1 (issue 1921): a merge held because the indexer cannot learn its own build is raised past the revision bound from its since, naming why", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.TakeInCaughtUp()
|
||||
body, err := takein.EncodeTakeIn(takein.TakeIn{Kind: takein.TakeInPending, Key: revKey, What: takein.TakeInRevision, Since: c.t,
|
||||
Merges: []takein.Merge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}, Unknown: "the engine does not answer"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w.TakeIn(takeInEvent(t, takein.TakeInPending, revKey), body)
|
||||
if o.ttls[signalKey("take-in", revKey)] != TakeInRevisionBound || len(o.findings) != 0 {
|
||||
t.Fatalf("an unknown own build: ttls %v, findings %v", o.ttls, o.findings)
|
||||
}
|
||||
expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound)
|
||||
for _, f := range o.findings {
|
||||
if !strings.Contains(f.Summary, "cannot learn the build it runs") {
|
||||
t.Errorf("finding %q does not say why", f.Summary)
|
||||
}
|
||||
}
|
||||
}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}},
|
||||
{"ADR 0633 §5 (1585/T21): a drop clears the wait, and the merges left under it said pending again with the same since are raised from it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.TakeInCaughtUp() // a live message, heard after the replay
|
||||
w.TakeIn(takeInEvent(t, takein.TakeInPending, revKey), takeInBody(t, takein.TakeInPending, revKey, t0))
|
||||
|
||||
@@ -72,6 +72,9 @@ func (w *Watchman) takeIn(kind, key string, t takein.TakeIn) {
|
||||
return
|
||||
}
|
||||
bound, what := TakeInRevisionBound, "a source revision waits for the indexer build "+t.WaitsFor
|
||||
if t.Unknown != "" {
|
||||
what = "a source revision is held because the indexer cannot learn the build it runs (" + t.Unknown + ")"
|
||||
}
|
||||
if t.What == takein.TakeInGroup {
|
||||
bound, what = TakeInGroupBound, "a delivery group is gathered and its end has not come"
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user