IX23: the indexer keeps no mesh-wide hold, serves no tool or view of modules, and hears no engine's manifest reads
Builds: design 47 G §196, §185, §364 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -1,83 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// holdsTool reads the indexer's mesh-wide holds (novox/hq ADR 0632 §2), asks one again now, or moves one by a person's
|
||||
// word: the asks go through the takeins state, as indexer_take_in's do, and the indexer answers there.
|
||||
func holdsTool(read stdio.KeptState, ask asker, now func() time.Time, acts meshhandactsclient.Recorder) stdio.Tool {
|
||||
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
|
||||
return stdio.Tool{Name: index.HoldsTool,
|
||||
Description: "The modules whose new manifest the indexer holds from the whole mesh (ADR 0632 §2): a manifest every node-engine " +
|
||||
"reads (the store holder's first) that needs a reader capability read mesh-wide (" + strings.Join(index.MeshWideReads, ", ") +
|
||||
") which not every counted node-engine states, present or away (ADR 0624 §1). Per module: what it needs, which nodes " +
|
||||
"lack what (or why no node is counted yet), the manifest the modules state keeps and the one held, the merge that " +
|
||||
"brought it and since when. The hold moves by itself once every counted node-engine states it (on manifest-reads-changed, " +
|
||||
"the assignments's change or a merge). With module and ask_again, the engines of the nodes that lack it are asked again " +
|
||||
"and the hold is weighed now; with module, release and why, it moves now by your word whatever the engines state (a hand " +
|
||||
"act, recorded through mesh-hand-acts, which names the nodes it strands: each node whose engine lacks the reading, an " +
|
||||
"away node from its return, reads the new entry without it and an engine before form:seat-reach finds no store). A hold " +
|
||||
"standing past its bound is the finding " + index.HoldStandingKind + ". indexer_engine_reads reads what each node's engine states.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{
|
||||
"module": map[string]any{"type": "string", "description": "one module; every hold when left out"},
|
||||
"ask_again": map[string]any{"type": "string", "description": "\"true\": ask the lacking nodes' engines again and weigh the hold now"},
|
||||
"release": map[string]any{"type": "string", "description": "\"true\": move the hold now by your word, whatever the engines state"},
|
||||
"why": map[string]any{"type": "string", "description": "why it is asked again or moved, in plain words"}}},
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
module := str(args, "module")
|
||||
release := str(args, "release") == "true"
|
||||
if release || str(args, "ask_again") == "true" {
|
||||
key, value, err := index.NewHoldAgain(module, str(args, "why"), stdio.CallerOf(args), release, now())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if release {
|
||||
what := "moved the mesh-wide hold of " + module + " by hand"
|
||||
if mk, err := meshcatalogclient.ModulesKey(module); err == nil {
|
||||
if e, err := read.Get(mk); err == nil && e != nil {
|
||||
if h, err := index.DecodeHold(e.Value); err == nil {
|
||||
what += ", " + index.Strands(h)
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := acts.Record(meshhandactsclient.Request{What: what,
|
||||
Why: str(args, "why"), Cause: index.HoldsTool, Verb: "hold-release"}); err != nil {
|
||||
return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so the hold was not moved: %w", err)
|
||||
}
|
||||
}
|
||||
a, err := ask(key, value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a.Result, nil
|
||||
}
|
||||
keys, err := read.Keys()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the holds state cannot be listed: %w", err)
|
||||
}
|
||||
return index.ListHolds(keys, func(key string) ([]byte, error) {
|
||||
e, err := read.Get(key)
|
||||
if err != nil || e == nil {
|
||||
return nil, err
|
||||
}
|
||||
return e.Value, nil
|
||||
}, module)
|
||||
}}
|
||||
}
|
||||
|
||||
// engineReadsTool lists, asks again and drops what each counted node's engine reads, as the indexer keeps it.
|
||||
func engineReadsTool(now func() time.Time) stdio.Tool {
|
||||
return enginereads.Tool(index.EngineReadsTool, "the indexer", enginereads.KeptState(stdio.State(index.EngineReadsState)), true, now)
|
||||
}
|
||||
@@ -301,8 +301,7 @@ func main() {
|
||||
}, 10*time.Second)...)
|
||||
tools = append(tools, index.TryCandidateTools(keptRead{stdio.State("candidates")}, keptRead{stdio.State(index.TryRecoveryState)}, asks, time.Now)...)
|
||||
tools = append(tools, configReadersTool(stdio.NewWatchedStates(), trunk.State("modules")), busStateTool(trunk.State), requirementsTool(keptSource),
|
||||
packageUsersTool(trunk.State), seatDefinitionTool(keptSource), outOfStepTool(keptSource),
|
||||
holdsTool(stdio.State(index.HoldsState), asks, time.Now, acts), engineReadsTool(time.Now))
|
||||
packageUsersTool(trunk.State), seatDefinitionTool(keptSource), outOfStepTool(keptSource))
|
||||
if err := stdio.Serve("", tools); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err)
|
||||
os.Exit(1)
|
||||
|
||||
@@ -88,7 +88,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
|
||||
// The states and the stream the controller makes from the indexer's manifest (old sections, ADR 0595 §11).
|
||||
for bucket, history := range map[string]uint8{index.PendingBucket: 1, index.VerdictsBucket: index.VerdictsHistory,
|
||||
index.TakenBucket: 1, index.PackagesBucket: 1,
|
||||
index.SeatDefinitionsBucket: 1, index.RequirementsBucket: 1, findingsBucket: 1, index.HoldsBucket: 1, index.UnjudgedBucket: 1} {
|
||||
index.SeatDefinitionsBucket: 1, index.RequirementsBucket: 1, findingsBucket: 1, index.UnjudgedBucket: 1} {
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: history}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -31,7 +31,6 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
@@ -62,37 +61,6 @@ type awaitedStore struct {
|
||||
name string
|
||||
}
|
||||
|
||||
// expiringKeys is index.Bounds on a state whose keys take a time to live (key-ttl in the manifest): a key armed expires
|
||||
// at its end, which its watcher hears as a purge; one standing keeps its end.
|
||||
type expiringKeys struct{ b *process.AwaitedBucket }
|
||||
|
||||
func (e expiringKeys) Arm(ctx context.Context, key string, ends time.Time) error {
|
||||
kv := e.b.KV()
|
||||
if kv == nil {
|
||||
return process.ErrNotYet
|
||||
}
|
||||
left := time.Until(ends)
|
||||
if left < time.Second {
|
||||
left = time.Second // the bus's least time to live
|
||||
}
|
||||
value, _ := json.Marshal(map[string]any{"bounds": "the hold of " + key, "ends": ends.UTC()})
|
||||
if _, err := kv.Create(ctx, key, value, jetstream.KeyTTL(left)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e expiringKeys) Disarm(ctx context.Context, key string) error {
|
||||
kv := e.b.KV()
|
||||
if kv == nil {
|
||||
return nil
|
||||
}
|
||||
if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s awaitedStore) store() (kvStore, error) {
|
||||
kv := s.b.KV()
|
||||
if kv == nil {
|
||||
@@ -492,48 +460,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the
|
||||
// assignments is settled first, so the toolchains settle through the gate already knowing it.
|
||||
gate := buildrun.NewToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf)
|
||||
// What each counted node's engine reads (ADR 0624 §1), for the mesh-wide hold of a manifest every node reads (ADR 0632
|
||||
// §2): asked through the node-engine seat's verb as the indexer's own account, for every node the assignments names,
|
||||
// kept in the state engine-reads once it is bound; each change weighs the holds again.
|
||||
reads := newEngineReads(calls.CallKeptOn, gate, func() { ix.KickHolds(ctx) }, logger.Printf)
|
||||
reads.keeper.Once = calls.CallOn // its fault keeps the last statement at once while the kept call asks on (ADR 0668 §3)
|
||||
gate.OnPlaced = reads.placed
|
||||
ix.Counted, ix.AskEngines = reads.counted, reads.ask
|
||||
// The holds, new with ADR 0632 (issue 1790's kind), bound when they appear: until then a merge that would hold a
|
||||
// manifest is left for redelivery, said, rather than write what a counted engine cannot read.
|
||||
holds := process.AwaitBucket(ctx, js, conn, index.HoldsBucket, "a merge whose manifest every node reads, needing what not every counted node-engine states, is left for redelivery", logger.Printf)
|
||||
ix.Holds = awaitedStore{b: holds}
|
||||
holds.Ready(func(jetstream.KeyValue) { ix.KickHolds(ctx) })
|
||||
// Each standing hold's bound, a key that expires at its end (ADR 0632 §2, the indexer's own state): its expiry weighs
|
||||
// the holds again, which raises manifest-held; until the state is bound a hold past its bound is raised at its next
|
||||
// weighing only, said.
|
||||
holdBounds := process.AwaitBucket(ctx, js, conn, index.HoldBoundsBucket, "a hold standing past its bound is raised at its next weighing only", logger.Printf)
|
||||
ix.HoldBounds = expiringKeys{holdBounds}
|
||||
// A kept manifest-reads call unanswered past five minutes, its node's machine heard (the watchdogs seat's presence,
|
||||
// asked when the bound ends), is call-unanswered in the indexer's findings (issue 1966, ADR 0668 §3, ADR 0596 §2).
|
||||
reads.callsOf(ctx, findings.KV, ix.HoldBounds, heardBy(calls.CallKept), logger.Printf)
|
||||
holdBounds.Ready(func(kv jetstream.KeyValue) {
|
||||
each := func(e jetstream.KeyValueEntry) {
|
||||
if node, ok := enginereads.BoundOf(e.Key()); ok {
|
||||
if e.Operation() == jetstream.KeyValuePurge { // a manifest-reads call's five minutes ended
|
||||
reads.expiredCall(node)
|
||||
}
|
||||
return
|
||||
}
|
||||
if e.Operation() == jetstream.KeyValuePurge { // its time to live ended
|
||||
ix.HoldBoundEnded(ctx, e.Key())
|
||||
}
|
||||
}
|
||||
if w, err := settle(ctx, kv, each); err != nil {
|
||||
process.NotWatched(logger.Printf, index.HoldBoundsBucket, err, "a hold standing past its bound is raised at its next weighing only until the indexer starts again")
|
||||
} else {
|
||||
go func() { errs <- follow(ctx, w, each) }()
|
||||
}
|
||||
ix.KickHolds(ctx) // a bound that ended while the indexer was down is raised now
|
||||
})
|
||||
process.AwaitBucket(ctx, js, conn, index.EngineReadsBucket, "what each node's engine reads is kept in memory only", logger.Printf).Ready(func(kv jetstream.KeyValue) {
|
||||
reads.bind(ctx, kv, errs, logger.Printf)
|
||||
})
|
||||
// The assignments is the assigner's: read through its first view and its views family, never its bucket (DLV-F1r).
|
||||
assignmentsErrs := make(chan error)
|
||||
// The indexer's one pull of its durable on the shared events stream (process.EventHub): every handler is registered
|
||||
@@ -681,17 +607,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// The versions each package's series holds (DLV-I5): the registry seats' published and the builder's numbered, on the
|
||||
// indexer's durable consumer of the shared events stream. Where it is not there yet, no series moves until the indexer
|
||||
// starts again, said.
|
||||
if err := hub.Handle(enginereads.SubjectPrefix+"*", func(m jetstream.Msg) { // the node-engine seat's manifest-reads-changed (ADR 0618 §4): its node asked again
|
||||
if !reads.heard(m, logger.Printf) {
|
||||
_ = m.Ack()
|
||||
}
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := hub.Handle(meshdelivery.SubjectSourceMovedAll, func(m jetstream.Msg) { source.heard(ctx, m) }); err != nil {
|
||||
return err
|
||||
}
|
||||
hub.OnBound(reads.atBind) // what a node's engine said before the consumer was bound (ADR 0668 §4)
|
||||
versions := func(m jetstream.Msg) {
|
||||
if !heardVersion(ctx, ix, m, logger.Printf) {
|
||||
_ = m.Ack() // nothing else the indexer consumes rides the shared stream
|
||||
|
||||
@@ -14,10 +14,6 @@ import (
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
@@ -87,65 +83,6 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part)
|
||||
if em := viewsEmitter(context.Background(), nil, process.Account{}, nil, &process.AwaitedBucket{}, nil, t.Logf); em.Findings == nil {
|
||||
t.Error("the indexer's views emitter writes no view-unsaid finding")
|
||||
}
|
||||
t.Run("the consumer bound: every counted node's manifest-reads called once", func(t *testing.T) { // 1966 note 155, ADR 0668 §4
|
||||
// A manifest-reads-changed said before the indexer's consumer existed is never delivered: the hub's OnBound
|
||||
// (reads.atBind) calls every node the assignments counts, so a stale statement ends at the bind.
|
||||
var called []string
|
||||
reads := &engineReads{known: true, nodes: []string{"ace", "g14"}, keeper: &enginereads.Keeper{Sync: true,
|
||||
Call: func(node, _ string, _ any) (json.RawMessage, error) {
|
||||
called = append(called, node)
|
||||
return nil, errors.New("no responders")
|
||||
}}}
|
||||
reads.atBind(time.Unix(0, 0))
|
||||
if strings.Join(called, ",") != "ace,g14" {
|
||||
t.Errorf("at the bind called %v, want ace and g14 once each", called)
|
||||
}
|
||||
})
|
||||
t.Run("a manifest-reads call unanswered past five minutes, its machine heard, is call-unanswered (ADR 0668 §3)", func(t *testing.T) {
|
||||
// A bound is kept at five minutes; when it ends the watchdogs seat's presence is asked, once (ADR 0596 §2): away,
|
||||
// no finding; heard, the finding in the indexer's findings, stamped with the wait's clock; answered, it is cleared.
|
||||
now := time.Unix(1000, 0)
|
||||
clock := func() time.Time { return now }
|
||||
down := true
|
||||
held := &findingsKV{vals: map[string][]byte{}}
|
||||
reads := &engineReads{known: true, nodes: []string{"ace"}}
|
||||
reads.keeper = &enginereads.Keeper{Sync: true, Now: clock, Call: func(node, _ string, _ any) (json.RawMessage, error) {
|
||||
if down {
|
||||
return nil, errors.New("no responders")
|
||||
}
|
||||
return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"form:new"}, StatedAt: now.Format(time.RFC3339Nano)})
|
||||
}}
|
||||
reads.keeper.Changed = reads.stepCalls
|
||||
armed := map[string]time.Time{}
|
||||
online, asked := false, []any{}
|
||||
heard := heardBy(func(key string, body any) (json.RawMessage, error) {
|
||||
asked = append(asked, key, body)
|
||||
return json.Marshal(meshwatchdogs.PresenceAnswer{Machine: "ace", Online: online})
|
||||
})
|
||||
calls := reads.callsOf(context.Background(), func() jetstream.KeyValue { return held }, nil, heard, t.Logf)
|
||||
calls.Now = clock
|
||||
calls.Arm = func(_ context.Context, node string, ends time.Time) error { armed[node] = ends; return nil }
|
||||
key := mcc.Key(mcc.ScopeCall, "manifest-reads.ace", mcc.KindCallUnanswered)
|
||||
reads.keeper.AskNode("ace", "counted") // unanswered at 1000
|
||||
if len(held.vals) != 0 || len(asked) != 0 || !armed["ace"].Equal(time.Unix(1000, 0).Add(mcc.CallUnansweredAfter)) {
|
||||
t.Fatalf("before five minutes: findings %v, presence asked %v, bounds %v", held.vals, asked, armed)
|
||||
}
|
||||
now = time.Unix(1000, 0).Add(mcc.CallUnansweredAfter + time.Second)
|
||||
reads.expiredCall("ace")
|
||||
if len(held.vals) != 0 || len(asked) != 2 || asked[0] != presenceInvoke || asked[1] != (meshwatchdogs.PresenceInput{Machine: "ace"}) {
|
||||
t.Fatalf("its machine away: findings %v, presence asked %v", held.vals, asked)
|
||||
}
|
||||
online = true
|
||||
reads.expiredCall("ace")
|
||||
if f, err := mcc.DecodeFinding(held.vals[key]); err != nil || f.Observer != "mesh-delivery-indexer" || f.Kind != mcc.KindCallUnanswered || !f.At.Equal(now) {
|
||||
t.Fatalf("heard five minutes on: finding %+v, %v; want it at the wait's clock %v", f, err, now)
|
||||
}
|
||||
down = false
|
||||
reads.keeper.AskNode("ace", "answered")
|
||||
if len(held.vals) != 0 || !reads.keeper.Statements()["ace"].Answered || len(asked) != 4 {
|
||||
t.Errorf("answered leaves %v, presence asked %v", held.vals, asked)
|
||||
}
|
||||
})
|
||||
// Milestone 47/G: the object buckets are the module's own by the name its account runs as, never a fixed name.
|
||||
if b := newBusObjects(nil, process.Account{Module: "renamed-indexer"}, index.CatalogRevisionsObjects).bucket(); b != "renamed-indexer_catalog-revisions" {
|
||||
t.Errorf("the catalog graph revisions open as %s, want the account's module's", b)
|
||||
|
||||
@@ -1,149 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"slices"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// engineReads is what each counted node's engine states it reads, for the indexer's mesh-wide hold (index.Holds; novox/hq
|
||||
// ADR 0632 §2, ADR 0624 §1), in the shape the builder keeps it: asked through the node-engine seat's verb manifest-reads
|
||||
// for every node the assignments names, on the indexer's start, on the seat's event manifest-reads-changed for that node and
|
||||
// when a person asks again (indexer_engine_reads, indexer_holds); kept in the indexer's state engine-reads, one key per
|
||||
// node, so a node that is away keeps its last statement. A node the assignments no longer names is forgotten. Until the
|
||||
// assignments is read no node is counted, and every hold stays.
|
||||
type engineReads struct {
|
||||
keeper *enginereads.Keeper
|
||||
gate *buildrun.ToolchainGate
|
||||
mu sync.Mutex
|
||||
known bool // the assignments is read
|
||||
nodes []string
|
||||
// calls is the wait of the kept calls (callsOf); nil weighs none. ctx is the indexer's.
|
||||
calls *enginereads.ReadsCalls
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
// newEngineReads asks through call (process.Conn.CallKeptOn, a kept call; ADR 0668) and tells changed after each statement changed.
|
||||
func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, changed func(), logf func(string, ...any)) *engineReads {
|
||||
r := &engineReads{gate: gate, ctx: context.Background()}
|
||||
r.keeper = &enginereads.Keeper{Call: call, Logf: logf, Changed: func() {
|
||||
if changed != nil {
|
||||
changed()
|
||||
}
|
||||
r.stepCalls()
|
||||
}}
|
||||
return r
|
||||
}
|
||||
|
||||
// stepCalls weighs the kept manifest-reads calls (enginereads.ReadsCalls): after a statement changed and the assignments's
|
||||
// change.
|
||||
func (r *engineReads) stepCalls() {
|
||||
if r.calls != nil {
|
||||
r.calls.Step(r.ctx)
|
||||
}
|
||||
}
|
||||
|
||||
// expiredCall weighs node's kept manifest-reads call when its bound ended (the key's expiry in the state hold-bounds).
|
||||
func (r *engineReads) expiredCall(node string) {
|
||||
if r.calls != nil {
|
||||
r.calls.Expired(r.ctx, node)
|
||||
}
|
||||
}
|
||||
|
||||
// bind keeps the statements in kv and hears it: every node's last statement at the start, a person's ask again and drop
|
||||
// later. Every counted node is asked again once bound, so what was asked before is kept in it too.
|
||||
func (r *engineReads) bind(ctx context.Context, kv jetstream.KeyValue, errs chan<- error, logf func(string, ...any)) {
|
||||
r.keeper.Put = func(node string, v []byte) error { _, err := kv.Put(ctx, node, v); return err }
|
||||
r.keeper.Delete = func(node string) error { return kv.Delete(ctx, node) }
|
||||
each := func(e jetstream.KeyValueEntry) {
|
||||
r.keeper.Heard(e.Key(), e.Value(), e.Operation() != jetstream.KeyValuePut)
|
||||
}
|
||||
w, err := settle(ctx, kv, each)
|
||||
if err != nil {
|
||||
process.NotWatched(logf, index.EngineReadsBucket, err, index.EngineReadsTool+" is not answered until the indexer starts again")
|
||||
return
|
||||
}
|
||||
go func() { errs <- follow(ctx, w, each) }()
|
||||
r.mu.Lock()
|
||||
nodes := slices.Clone(r.nodes)
|
||||
r.mu.Unlock()
|
||||
r.ask(nodes, "the state engine-reads is bound")
|
||||
}
|
||||
|
||||
// placed takes the assignments in, once it is read and after each change of it: a node it names newly is asked, one it no
|
||||
// longer names is forgotten, key and all.
|
||||
func (r *engineReads) placed() {
|
||||
nodes := r.gate.Nodes()
|
||||
r.mu.Lock()
|
||||
before, first := r.nodes, !r.known
|
||||
r.nodes, r.known = nodes, true
|
||||
r.mu.Unlock()
|
||||
for _, node := range nodes {
|
||||
if first || !slices.Contains(before, node) {
|
||||
r.keeper.AskNode(node, "counted: the assignments names it")
|
||||
}
|
||||
}
|
||||
for _, node := range before {
|
||||
if !slices.Contains(nodes, node) {
|
||||
r.keeper.Forget(node)
|
||||
}
|
||||
}
|
||||
if r.keeper.Changed != nil {
|
||||
r.keeper.Changed()
|
||||
}
|
||||
r.stepCalls()
|
||||
}
|
||||
|
||||
// atBind asks every counted node again when the indexer's consumer is bound (the hub's OnBound): a manifest-reads-changed
|
||||
// said before the consumer existed is not delivered to it (novox/hq ADR 0668 §4; issue 1966 note 155).
|
||||
func (r *engineReads) atBind(created time.Time) {
|
||||
r.mu.Lock()
|
||||
nodes := slices.Clone(r.nodes)
|
||||
r.mu.Unlock()
|
||||
r.keeper.AtBind(func() []string { return nodes })(created)
|
||||
}
|
||||
|
||||
// heard takes an event of the indexer's consumer on EVENTS: manifest-reads-changed asks its node again, and answers true.
|
||||
func (r *engineReads) heard(m jetstream.Msg, logf func(string, ...any)) bool {
|
||||
node, ours, err := enginereads.NodeOf(m.Subject(), m.Data())
|
||||
if !ours {
|
||||
return false
|
||||
}
|
||||
if ackErr := m.Ack(); ackErr != nil {
|
||||
logf("acknowledging %s: %v", m.Subject(), ackErr)
|
||||
}
|
||||
if err != nil {
|
||||
logf("%s holds what is not %s, and %s is asked again all the same: %v", m.Subject(), enginereads.Event, node, err)
|
||||
}
|
||||
r.keeper.AskNode(node, enginereads.Event)
|
||||
return true
|
||||
}
|
||||
|
||||
// counted is index.Indexer.Counted: what each counted node states (nil, the baseline, for one with no statement), and
|
||||
// whether the assignments is read.
|
||||
func (r *engineReads) counted() (map[string][]string, bool) {
|
||||
r.mu.Lock()
|
||||
nodes, known := slices.Clone(r.nodes), r.known
|
||||
r.mu.Unlock()
|
||||
if !known {
|
||||
return nil, false
|
||||
}
|
||||
return r.keeper.Reads(nodes), true
|
||||
}
|
||||
|
||||
// ask asks these nodes' engines again (index.Indexer.AskEngines).
|
||||
func (r *engineReads) ask(nodes []string, why string) {
|
||||
for _, node := range nodes {
|
||||
r.keeper.AskNode(node, why)
|
||||
}
|
||||
}
|
||||
@@ -1,82 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// presenceInvoke is the watchdogs seat's verb presence as the manifest's invokes names it: whether a node's machine is
|
||||
// heard, asked of the seat, never read from its holder's state (novox/hq ADR 0596 §2).
|
||||
const presenceInvoke = "seat:" + meshwatchdogs.Seat + "." + meshwatchdogs.VerbPresence
|
||||
|
||||
// heardBy asks the watchdogs seat's presence of one machine through call (process.Conn.CallKept, a read) and reads its
|
||||
// declared answer: whether the machine is heard now.
|
||||
func heardBy(call func(key string, body any) (json.RawMessage, error)) func(context.Context, string) (bool, error) {
|
||||
return func(_ context.Context, machine string) (bool, error) {
|
||||
raw, err := call(presenceInvoke, meshwatchdogs.PresenceInput{Machine: machine})
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
var a meshwatchdogs.PresenceAnswer
|
||||
if err := json.Unmarshal(raw, &a); err != nil {
|
||||
return false, fmt.Errorf("%s does not read: %w", presenceInvoke, err)
|
||||
}
|
||||
return a.Online, nil
|
||||
}
|
||||
}
|
||||
|
||||
// callsOf is the wait of the indexer's kept manifest-reads calls (enginereads.ReadsCalls): a node whose engine has not
|
||||
// answered for five minutes is, when its bound ends and the watchdogs seat says its machine is heard, the finding
|
||||
// call-unanswered in the indexer's findings, stamped with the wait's clock (issue 1966, ADR 0668 §3); its bound is a key
|
||||
// of the state hold-bounds, which expires at its end.
|
||||
func (r *engineReads) callsOf(ctx context.Context, findings func() jetstream.KeyValue, bounds index.Bounds, heard func(context.Context, string) (bool, error), logf func(string, ...any)) *enginereads.ReadsCalls {
|
||||
c := &enginereads.ReadsCalls{Keeper: r.keeper, Heard: heard, Bound: mcc.CallUnansweredAfter, Logf: logf,
|
||||
Nodes: func() []string { r.mu.Lock(); defer r.mu.Unlock(); return slices.Clone(r.nodes) },
|
||||
Raise: func(ctx context.Context, u enginereads.UnansweredCall) error {
|
||||
kv := findings()
|
||||
if kv == nil {
|
||||
return process.ErrNotYet
|
||||
}
|
||||
o := mcc.CallUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "indexer."+mcc.KindCallUnanswered, u.Since)
|
||||
o.Needs = index.EngineReadsTool + " with ask_again asks the node again now and shows its statement."
|
||||
value, err := mcc.EncodeFinding(o.Finding(index.Module, "", u.At))
|
||||
if err == nil {
|
||||
_, err = kv.Put(ctx, o.Key(), value)
|
||||
}
|
||||
return err
|
||||
},
|
||||
Clear: func(ctx context.Context, node string) error {
|
||||
kv := findings()
|
||||
if kv == nil {
|
||||
return nil
|
||||
}
|
||||
key := mcc.Key(mcc.ScopeCall, enginereads.CallID(node), mcc.KindCallUnanswered)
|
||||
if _, err := kv.Get(ctx, key); err != nil {
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
return kv.Delete(ctx, key)
|
||||
}}
|
||||
if bounds != nil {
|
||||
c.Arm = func(ctx context.Context, node string, ends time.Time) error {
|
||||
return bounds.Arm(ctx, enginereads.BoundPrefix+node, ends)
|
||||
}
|
||||
}
|
||||
r.ctx, r.calls = ctx, c
|
||||
return c
|
||||
}
|
||||
@@ -19,7 +19,6 @@ type takerIn interface {
|
||||
JudgeAgain(ctx context.Context, module, digest string) index.TakenIn
|
||||
SayAcceptedAgain(ctx context.Context, module string) index.TakenIn
|
||||
SeatDefinitionsAgain(ctx context.Context, module string, pull int, head string) index.TakenIn
|
||||
HoldAgain(ctx context.Context, module string, release bool, by string) index.TakenIn
|
||||
TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn
|
||||
DropPending(ctx context.Context, key, by, why string, abandon bool) index.TakenIn
|
||||
TakeInMerge(ctx context.Context, repository, merge string, seq uint64, by, why string) index.TakenIn
|
||||
@@ -75,12 +74,6 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
|
||||
} else if a.Repository != "" {
|
||||
asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository)
|
||||
a.Module = a.Repository
|
||||
} else if a.Hold != "" {
|
||||
asked = "weighing of its hold again"
|
||||
if a.Hold == index.HoldRelease {
|
||||
asked = "move of its hold by hand"
|
||||
}
|
||||
r = ix.HoldAgain(ctx, a.Module, a.Hold == index.HoldRelease, a.By)
|
||||
} else if a.SeatDefinitions {
|
||||
asked, r = "reading of its seat definitions again", ix.SeatDefinitionsAgain(ctx, a.Module, a.Pull, a.Head)
|
||||
} else if a.Say {
|
||||
|
||||
@@ -137,19 +137,6 @@ type Indexer struct {
|
||||
// that finds the candidates of another pull request asks it and deletes those of a pull request no longer open, so a
|
||||
// closed message the indexer missed heals at the next one (issue 1126); nil asks none.
|
||||
ReadOpen func(ctx context.Context) ([]forgemsg.OpenPull, error)
|
||||
// Holds is the holds state (holds.go; novox/hq ADR 0632 §2): a module's new entry whose manifest needs a reader
|
||||
// capability read mesh-wide, waiting for every counted node-engine to state it. Nil, or Counted nil, holds nothing.
|
||||
Holds Store
|
||||
// Counted is what each counted node's engine states it reads (ADR 0624 §1; nil, the baseline, for a node with no
|
||||
// statement), and whether the counted nodes are known (the assignments is read).
|
||||
Counted func() (map[string][]string, bool)
|
||||
// AskEngines asks these nodes' engines what they read again (the node-engine seat's verb manifest-reads); nil asks none.
|
||||
AskEngines func(nodes []string, why string)
|
||||
holding atomic.Bool // a MoveHolds runs in the background (KickHolds)
|
||||
holdAgain atomic.Bool // a kick arrived while it ran
|
||||
// HoldBounds is the indexer's state hold-bounds (holds.go): each standing hold's bound as a key that expires at its
|
||||
// end, whose expiry weighs the holds again (HoldBoundEnded). Nil raises the finding at the next weighing only.
|
||||
HoldBounds Bounds
|
||||
|
||||
mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
|
||||
source sourceNow
|
||||
|
||||
@@ -1,565 +0,0 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"reflect"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/graph"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/trunk"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// The indexer's mesh-wide hold (novox/hq ADR 0632 §2, ADR 0624 §1). Every node-engine reads the store holder's manifest
|
||||
// wherever the store runs, to find the store (mesh-host internal/compose/frombus), and an engine before the reading
|
||||
// form:seat-reach finds it only through a module serving the retired provision artifact-store. A manifest that needs a
|
||||
// reader capability read mesh-wide (MeshWideReads) is therefore not written to a module's entry in the modules state
|
||||
// while some counted node-engine does not state it: the entry keeps its previous manifest, fingerprint and inputs, and
|
||||
// the new one waits in the state holds, one key per module, until every counted node-engine states what it needs. Then
|
||||
// it is written as its merge would have written it, judged and said. The assigner's hold of ADR 0614 §4 holds a want on
|
||||
// one node; this one holds the entry itself, which every node reads.
|
||||
//
|
||||
// "Counted" is ADR 0624 §1's set: every node the assignments names, present or away, each read as its engine last stated
|
||||
// through the node-engine seat's verb manifest-reads (a node with no statement read as the baseline; the indexer's state
|
||||
// engine-reads, nodeengineclient/enginereads). Until the assignments is read no node is counted and every hold stays. The hold moves
|
||||
// on the event manifest-reads-changed, the assignments's change, a person's ask again, and a merge; never on a timer.
|
||||
const (
|
||||
// HoldsState is the indexer's state of held entries: the bucket mesh-delivery-indexer_holds, keyed as the modules state.
|
||||
HoldsState = "holds"
|
||||
HoldsBucket = Module + "_" + HoldsState
|
||||
// HoldsTool reads the holds, asks one again now, or moves one by a person's word.
|
||||
HoldsTool = "indexer_holds"
|
||||
// EngineReadsState is the indexer's state of what each counted node's engine states it reads (nodeengineclient/enginereads).
|
||||
EngineReadsState = "engine-reads"
|
||||
EngineReadsBucket = Module + "_" + EngineReadsState
|
||||
// EngineReadsTool lists, asks again and drops those statements.
|
||||
EngineReadsTool = "indexer_engine_reads"
|
||||
)
|
||||
|
||||
// MeshWideReads are the reader capabilities whose need holds a module's new entry from the whole mesh, not only from the
|
||||
// nodes it is placed on: form:seat-reach, which catalogue.ManifestNeeds marks for the store holder's own manifest (a
|
||||
// claim of container-image-registry at the mesh's scope) and for a manifest naming a seat's reach (ADR 0632 §1, §2).
|
||||
var MeshWideReads = []string{catalogue.ReadForm + catalogue.ReferenceFormSeatReach}
|
||||
|
||||
// A hold standing past HoldBound is the indexer's finding HoldStandingKind about its module, which the condition keeper
|
||||
// raises (ADR 0564 §2): the module's new entry is delivered nowhere while it stands, and a counted node that is away for
|
||||
// long (ADR 0624 §1) keeps it so. The finding names who lacks what; it is cleared when the hold moves or is dropped.
|
||||
// The bound is a key of the indexer's state hold-bounds that expires at Since+HoldBound (a key's time to live on the
|
||||
// bus, as the builder's bounds): its expiry weighs the holds again, which raises the finding, and a restarted indexer
|
||||
// weighs every hold at its start, so a bound that ended while it was down is raised then. No timer runs.
|
||||
const (
|
||||
HoldStandingKind = "manifest-held"
|
||||
HoldBound = 30 * time.Minute
|
||||
HoldBoundsState = "hold-bounds"
|
||||
HoldBoundsBucket = Module + "_" + HoldBoundsState
|
||||
)
|
||||
|
||||
// Bounds is the state hold-bounds: Arm writes key to expire at ends, a standing key keeping its end; Disarm deletes it.
|
||||
type Bounds interface {
|
||||
Arm(ctx context.Context, key string, ends time.Time) error
|
||||
Disarm(ctx context.Context, key string) error
|
||||
}
|
||||
|
||||
// HoldVersion is the version of a holds value this indexer writes and reads.
|
||||
const HoldVersion = 1
|
||||
|
||||
// Hold is one module's held entry: what the new manifest needs of MeshWideReads, which counted nodes lack which of it
|
||||
// (or why no node is counted), the manifest the modules state keeps and the one held, the merge that brought it, since
|
||||
// when, and the entry written once it moves.
|
||||
type Hold struct {
|
||||
V int `json:"v"`
|
||||
Module string `json:"module"`
|
||||
Needs []string `json:"needs"`
|
||||
Lacking map[string][]string `json:"lacking,omitempty"`
|
||||
Unknown string `json:"unknown,omitempty"`
|
||||
Kept string `json:"kept"`
|
||||
Manifest string `json:"manifest"`
|
||||
Repository string `json:"repository"`
|
||||
Commit string `json:"commit"`
|
||||
Since time.Time `json:"since"`
|
||||
Next json.RawMessage `json:"next"`
|
||||
}
|
||||
|
||||
// EncodeHold is h at HoldVersion.
|
||||
func EncodeHold(h Hold) ([]byte, error) {
|
||||
if h.Module == "" || len(h.Next) == 0 {
|
||||
return nil, errors.New("a hold names its module and the entry it holds")
|
||||
}
|
||||
h.V = HoldVersion
|
||||
return json.Marshal(h)
|
||||
}
|
||||
|
||||
// DecodeHold reads a holds value; it refuses another version and one holding no entry.
|
||||
func DecodeHold(raw []byte) (Hold, error) {
|
||||
var h Hold
|
||||
if err := json.Unmarshal(raw, &h); err != nil {
|
||||
return h, err
|
||||
}
|
||||
if h.V != HoldVersion {
|
||||
return h, fmt.Errorf("a holds value of version %d, not %d", h.V, HoldVersion)
|
||||
}
|
||||
if h.Module == "" || len(h.Next) == 0 {
|
||||
return h, errors.New("a holds value naming no module or no entry")
|
||||
}
|
||||
return h, nil
|
||||
}
|
||||
|
||||
// meshWide is what of needs MeshWideReads names, sorted.
|
||||
func meshWide(needs []string) []string {
|
||||
var out []string
|
||||
for _, n := range needs {
|
||||
if slices.Contains(MeshWideReads, n) {
|
||||
out = append(out, n)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return slices.Compact(out)
|
||||
}
|
||||
|
||||
// noneCounted is why a hold stays while no node is counted.
|
||||
const noneCounted = "no node is counted yet (the assignments is not read), so the hold stays until it is"
|
||||
|
||||
// lacking is, of the counted nodes, each whose engine lacks some of needs, with what it lacks; or why no node is counted.
|
||||
// It judges as the assigner does (graph.MissingReads): a bootstrapped seat is read by every engine.
|
||||
func (ix *Indexer) lacking(needs []string) (map[string][]string, string) {
|
||||
counted, known := ix.Counted()
|
||||
if !known || len(counted) == 0 {
|
||||
return nil, noneCounted
|
||||
}
|
||||
out := map[string][]string{}
|
||||
for node, stated := range counted {
|
||||
if miss := graph.MissingReads(needs, stated); len(miss) > 0 {
|
||||
out[node] = miss
|
||||
}
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return nil, ""
|
||||
}
|
||||
return out, ""
|
||||
}
|
||||
|
||||
// lackingWords says who lacks what, or why no node is counted.
|
||||
func lackingWords(lacking map[string][]string, unknown string) string {
|
||||
if unknown != "" {
|
||||
return unknown
|
||||
}
|
||||
var parts []string
|
||||
for _, node := range slices.Sorted(maps.Keys(lacking)) {
|
||||
parts = append(parts, node+" lacks "+strings.Join(lacking[node], ", "))
|
||||
}
|
||||
return strings.Join(parts, "; ")
|
||||
}
|
||||
|
||||
// holdBack decides, for one module of a merge, whether its new entry next waits (true: kept in the holds state, the
|
||||
// modules state left at cur) or is written now (false: any hold of it dropped, the merge having moved it). It holds only
|
||||
// a changed manifest of a module the modules state holds an entry of: a new module has no previous entry to keep, and is
|
||||
// held on each node by the assigner (ADR 0614 §4). An error is a hold that could not be kept: the merge is left, and
|
||||
// delivered again, rather than writing what a counted engine cannot read.
|
||||
func (ix *Indexer) holdBack(ctx context.Context, key string, next, cur meshcatalog.ModulesByModuleValue, has bool) (bool, error) {
|
||||
if ix.Holds == nil || ix.Counted == nil {
|
||||
return false, nil
|
||||
}
|
||||
needs := meshWide(next.Needs)
|
||||
var lacking map[string][]string
|
||||
var unknown string
|
||||
// Any change of the entry is held, not only a new manifest: a template's text that newly needs a reading changes
|
||||
// the module's tree and needs while its manifest stays (ADR 0632 §2, docker's daemon file).
|
||||
if len(needs) > 0 && has && cur.Manifest != "" && next.Manifest != "" && !sameButWhere(cur, next) {
|
||||
lacking, unknown = ix.lacking(needs)
|
||||
}
|
||||
if len(lacking) == 0 && unknown == "" {
|
||||
return false, ix.dropHold(ctx, key, "its merge "+next.Commit+" brought a manifest every counted node-engine reads")
|
||||
}
|
||||
raw, err := meshcatalogclient.EncodeModule(next)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
h := Hold{Module: key, Needs: needs, Lacking: lacking, Unknown: unknown, Kept: cur.Manifest, Manifest: next.Manifest,
|
||||
Repository: next.Repository, Commit: next.Commit, Since: ix.now().UTC(), Next: raw}
|
||||
err = write(ctx, ix.Holds, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
h.Since, h.Next, h.Commit = ix.now().UTC(), raw, next.Commit
|
||||
if found {
|
||||
if old, err := DecodeHold(value); err == nil {
|
||||
// The module is held since its first hold, whatever a later merge brings; the same entry held again keeps
|
||||
// its first merge too, as a module's commit does (writeOverlay).
|
||||
h.Since = old.Since
|
||||
if old.Manifest == h.Manifest && sameHeld(old.Next, next) {
|
||||
h.Next, h.Commit = old.Next, old.Commit
|
||||
}
|
||||
}
|
||||
}
|
||||
data, err := EncodeHold(h)
|
||||
return data, false, err
|
||||
})
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("%s: its new manifest %s is held from the mesh, and the hold cannot be kept: %w", key, next.Manifest, err)
|
||||
}
|
||||
ix.logf("modules.%s keeps manifest %s: the new manifest %s of %s needs %s, which not every counted node-engine states (%s; ADR 0632 §2); it moves once each does",
|
||||
key, cur.Manifest, next.Manifest, next.Commit, strings.Join(needs, ", "), lackingWords(lacking, unknown))
|
||||
ix.holdStands(ctx, key, h)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// dropHold deletes the hold of key, if there is one, saying why.
|
||||
func (ix *Indexer) dropHold(ctx context.Context, key, why string) error {
|
||||
if ix.Holds == nil {
|
||||
return nil
|
||||
}
|
||||
dropped := false
|
||||
err := write(ctx, ix.Holds, key, func(_ []byte, found bool) ([]byte, bool, error) {
|
||||
dropped = found
|
||||
return nil, found, nil
|
||||
})
|
||||
if errors.Is(err, process.ErrNotYet) {
|
||||
return nil // a holds state not on the bus yet holds nothing to drop
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("holds.%s: %w", key, err)
|
||||
}
|
||||
if dropped {
|
||||
ix.logf("holds.%s dropped: %s", key, why)
|
||||
ix.holdEnds(ctx, key)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// KickHolds moves every hold whose needs every counted node-engine now states, in the background, once at a time: on a
|
||||
// node's statement changed, the assignments's change and the holds state's binding.
|
||||
func (ix *Indexer) KickHolds(ctx context.Context) {
|
||||
if ix.Holds == nil || ix.Counted == nil {
|
||||
return
|
||||
}
|
||||
if ix.holding.Swap(true) {
|
||||
ix.holdAgain.Store(true)
|
||||
return
|
||||
}
|
||||
go func() {
|
||||
for {
|
||||
ix.holdAgain.Store(false)
|
||||
if err := ix.MoveHolds(ctx); err != nil {
|
||||
ix.logf("moving the holds: %v", err)
|
||||
}
|
||||
if !ix.holdAgain.Load() {
|
||||
ix.holding.Store(false)
|
||||
if !ix.holdAgain.Load() || ix.holding.Swap(true) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// MoveHolds weighs every hold against what the counted node-engines state now: one every node reads is written, and
|
||||
// one still held is kept with who lacks what now.
|
||||
func (ix *Indexer) MoveHolds(ctx context.Context) error {
|
||||
if ix.Holds == nil || ix.Counted == nil {
|
||||
return nil
|
||||
}
|
||||
keys, err := ix.Holds.Keys(ctx)
|
||||
if errors.Is(err, process.ErrNotYet) {
|
||||
return nil // weighed once the state is bound (main.go)
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("listing the holds: %w", err)
|
||||
}
|
||||
var errs []error
|
||||
for _, key := range slices.Sorted(slices.Values(keys)) {
|
||||
if _, _, err := ix.moveHold(ctx, key, ""); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// moveHold weighs one hold; with by, a person's word moves it whatever the engines state. It answers whether it moved,
|
||||
// and what stands.
|
||||
func (ix *Indexer) moveHold(ctx context.Context, key, by string) (bool, string, error) {
|
||||
moved := false
|
||||
for try := 0; try < writeTries; try++ {
|
||||
value, rev, found, err := ix.Holds.Get(ctx, key)
|
||||
if err != nil {
|
||||
return moved, "", fmt.Errorf("holds.%s: %w", key, err)
|
||||
}
|
||||
if !found {
|
||||
if moved {
|
||||
return true, "moved", nil
|
||||
}
|
||||
return false, "no hold of " + key + " is kept", nil
|
||||
}
|
||||
h, err := DecodeHold(value)
|
||||
if err != nil {
|
||||
ix.logf("holds.%s holds what is not a hold, left: %v", key, err)
|
||||
return moved, "", fmt.Errorf("holds.%s: %w", key, err)
|
||||
}
|
||||
next, err := meshcatalogclient.DecodeModule(h.Next)
|
||||
if err != nil {
|
||||
return moved, "", fmt.Errorf("holds.%s: its entry: %w", key, err)
|
||||
}
|
||||
lacking, unknown := ix.lacking(h.Needs)
|
||||
if by == "" && (len(lacking) > 0 || unknown != "") {
|
||||
if reflect.DeepEqual(lacking, h.Lacking) && unknown == h.Unknown {
|
||||
ix.holdStands(ctx, key, h)
|
||||
return false, lackingWords(lacking, unknown), nil
|
||||
}
|
||||
h.Lacking, h.Unknown = lacking, unknown
|
||||
data, err := EncodeHold(h)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
if err := ix.Holds.Update(ctx, key, data, rev); errors.Is(err, ErrConflict) {
|
||||
continue
|
||||
} else if err != nil {
|
||||
return false, "", fmt.Errorf("holds.%s: %w", key, err)
|
||||
}
|
||||
ix.holdStands(ctx, key, h)
|
||||
return false, lackingWords(lacking, unknown), nil
|
||||
}
|
||||
ix.recompute.Lock()
|
||||
// Under the lock a merge writes its modules by, read again: a merge since replaced or dropped the hold.
|
||||
if _, now, still, err := ix.Holds.Get(ctx, key); err != nil || !still || now != rev {
|
||||
ix.recompute.Unlock()
|
||||
if err != nil {
|
||||
return moved, "", fmt.Errorf("holds.%s: %w", key, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
curRaw, _, has, err := ix.Modules.Get(ctx, key)
|
||||
if err != nil {
|
||||
ix.recompute.Unlock()
|
||||
return moved, "", fmt.Errorf("modules.%s: %w", key, err)
|
||||
}
|
||||
cur, derr := meshcatalogclient.DecodeModule(curRaw)
|
||||
if !has || derr != nil || !trunk.SameRepository(cur.Repository, next.Repository) {
|
||||
err := ix.Holds.Delete(ctx, key, rev)
|
||||
ix.recompute.Unlock()
|
||||
if errors.Is(err, ErrConflict) {
|
||||
continue
|
||||
}
|
||||
if err == nil {
|
||||
ix.holdEnds(ctx, key)
|
||||
}
|
||||
ix.logf("holds.%s dropped: the modules state no longer holds the module of %s", key, next.Repository)
|
||||
return moved, "dropped: the modules state no longer holds the module of " + next.Repository, err
|
||||
}
|
||||
next.Src = int64(ix.srcOf(next.Repository, next.Commit))
|
||||
w := &mergeWrite{repository: next.Repository, merge: next.Commit}
|
||||
err = ix.recomputeLocked(ctx, map[string]meshcatalog.ModulesByModuleValue{key: next}, w)
|
||||
if err == nil {
|
||||
err = ix.Holds.Delete(ctx, key, rev)
|
||||
}
|
||||
ix.recompute.Unlock()
|
||||
if errors.Is(err, ErrConflict) {
|
||||
moved = true
|
||||
continue // a merge held a newer manifest meanwhile: weighed in turn
|
||||
}
|
||||
if err != nil {
|
||||
return moved, "", fmt.Errorf("moving the hold of %s: %w", key, err)
|
||||
}
|
||||
moved = true
|
||||
ix.holdEnds(ctx, key)
|
||||
why := "every counted node-engine states " + strings.Join(h.Needs, ", ")
|
||||
if by != "" {
|
||||
why = "moved by " + by + "'s word, " + stranded(lacking, unknown, h.Needs)
|
||||
}
|
||||
ix.logf("modules.%s moves from manifest %s to %s of %s, held since %s: %s", key, h.Kept, h.Manifest, h.Commit, h.Since.Format(time.RFC3339), why)
|
||||
if err := ix.RecomputeCandidates(ctx); err != nil {
|
||||
ix.logf("the candidates after the hold of %s moved: %v", key, err)
|
||||
}
|
||||
if ix.Verdicts != nil {
|
||||
if _, err := ix.JudgeAll(ctx, w.fresh); err != nil {
|
||||
ix.logf("judge: the manifests are not judged after the hold of %s moved: %v", key, err)
|
||||
}
|
||||
}
|
||||
return true, why, nil
|
||||
}
|
||||
return moved, "", fmt.Errorf("holds.%s: it kept changing while it was weighed", key)
|
||||
}
|
||||
|
||||
// sameHeld says a held entry is next but for where it came from: its first merge stands.
|
||||
func sameHeld(held json.RawMessage, next meshcatalog.ModulesByModuleValue) bool {
|
||||
prev, err := meshcatalogclient.DecodeModule(held)
|
||||
return err == nil && sameButWhere(prev, next)
|
||||
}
|
||||
|
||||
// holdStands keeps the finding of h standing: raised, naming who lacks what now, once h stood HoldBound; until then its
|
||||
// bound is armed in the state hold-bounds, whose expiry weighs the holds again (HoldBoundEnded).
|
||||
func (ix *Indexer) holdStands(ctx context.Context, key string, h Hold) {
|
||||
if ends := h.Since.Add(HoldBound); ix.now().Before(ends) {
|
||||
if ix.HoldBounds != nil {
|
||||
if err := ix.HoldBounds.Arm(ctx, key, ends); err != nil {
|
||||
ix.logf("hold-bounds.%s is not armed, so the hold standing past %s is raised at its next weighing only: %v", key, HoldBound, err)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
ix.finding(ctx, findingID("hold", key), HoldStandingKind, true,
|
||||
fmt.Sprintf("modules.%s keeps manifest %s since %s: its new manifest %s of %s needs %s, and %s",
|
||||
key, h.Kept, h.Since.Format(time.RFC3339), h.Manifest, cmpOr(h.Commit, "its merge"), strings.Join(h.Needs, ", "), lackingWords(h.Lacking, h.Unknown)),
|
||||
"A module's new manifest is held from the whole mesh",
|
||||
"Every node-engine reads "+key+"'s manifest, and one counted node-engine does not yet state the reading the new one "+
|
||||
"needs, so the indexer keeps the old entry and the new manifest is delivered nowhere (ADR 0632 §2). It moves by "+
|
||||
"itself once each counted node's engine states it; an away node keeps it held until it is back. indexer_holds "+
|
||||
"names who lacks what, asks again, or moves it by a person's word; indexer_engine_reads reads the statements.",
|
||||
"The hold of "+key+" moved or was dropped.")
|
||||
}
|
||||
|
||||
// HoldBoundEnded takes a key of the state hold-bounds that expired: the holds are weighed again, and the one it bounds,
|
||||
// still standing, is raised.
|
||||
func (ix *Indexer) HoldBoundEnded(ctx context.Context, key string) {
|
||||
ix.logf("hold-bounds.%s ended: the holds are weighed again", key)
|
||||
ix.KickHolds(ctx)
|
||||
}
|
||||
|
||||
// holdEnds disarms the bound of key's hold and clears its finding.
|
||||
func (ix *Indexer) holdEnds(ctx context.Context, key string) {
|
||||
if ix.HoldBounds != nil {
|
||||
if err := ix.HoldBounds.Disarm(ctx, key); err != nil {
|
||||
ix.logf("hold-bounds.%s is not deleted; its end weighs the holds once more, which moves nothing: %v", key, err)
|
||||
}
|
||||
}
|
||||
ix.finding(ctx, findingID("hold", key), HoldStandingKind, false, "", "", "", "")
|
||||
}
|
||||
|
||||
// Strands is what moving h by a person's word strands now, as its hand act and its answer say it.
|
||||
func Strands(h Hold) string { return stranded(h.Lacking, h.Unknown, h.Needs) }
|
||||
|
||||
// stranded says, of a hold moved by a person's word, the nodes it strands: each counted node whose engine lacks what
|
||||
// the new entry needs reads it without it (an engine before form:seat-reach finds no store) until it states it.
|
||||
func stranded(lacking map[string][]string, unknown string, needs []string) string {
|
||||
switch {
|
||||
case unknown != "":
|
||||
return "while " + unknown + ": every node whose engine lacks " + strings.Join(needs, ", ") + " is stranded until it states it"
|
||||
case len(lacking) == 0:
|
||||
return "every counted node-engine stating " + strings.Join(needs, ", ") + ": it strands none"
|
||||
}
|
||||
return "and it strands " + strings.Join(slices.Sorted(maps.Keys(lacking)), ", ") + " (" + lackingWords(lacking, "") +
|
||||
"): each reads the new entry without that reading until its engine states it, an away node from its return"
|
||||
}
|
||||
|
||||
func cmpOr(a, b string) string {
|
||||
if a != "" {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// HoldAgain answers a person's ask of one hold (the tool indexer_holds): again asks the engines of the counted nodes that
|
||||
// lack what it needs once more and weighs it now; by moves it by that person's word, whatever the engines state.
|
||||
func (ix *Indexer) HoldAgain(ctx context.Context, module string, release bool, by string) TakenIn {
|
||||
r := TakenIn{Module: module}
|
||||
key, err := meshcatalogclient.ModulesKey(module)
|
||||
if err != nil {
|
||||
r.Answer = err.Error()
|
||||
return r
|
||||
}
|
||||
if ix.Holds == nil || ix.Counted == nil {
|
||||
r.Answer = "the indexer keeps no holds: its holds state or the engines' statements are not bound"
|
||||
return r
|
||||
}
|
||||
value, _, found, err := ix.Holds.Get(ctx, key)
|
||||
switch {
|
||||
case err != nil:
|
||||
r.Answer = "the holds state cannot be read: " + err.Error()
|
||||
return r
|
||||
case !found:
|
||||
r.Answer = "no hold of " + module + " is kept: its entry in the modules state is the newest manifest merged"
|
||||
return r
|
||||
}
|
||||
h, err := DecodeHold(value)
|
||||
if err != nil {
|
||||
r.Answer = "its hold cannot be read: " + err.Error()
|
||||
return r
|
||||
}
|
||||
r.Repository, r.Commit = h.Repository, h.Commit
|
||||
if !release {
|
||||
lacking, _ := ix.lacking(h.Needs)
|
||||
if ix.AskEngines != nil && len(lacking) > 0 {
|
||||
ix.AskEngines(slices.Sorted(maps.Keys(lacking)), "the hold of "+module+" asked again")
|
||||
}
|
||||
}
|
||||
who := ""
|
||||
if release {
|
||||
who = cmpOr(by, "a person")
|
||||
}
|
||||
moved, words, err := ix.moveHold(ctx, key, who)
|
||||
switch {
|
||||
case err != nil:
|
||||
r.Answer = "the hold was not weighed: " + err.Error()
|
||||
case moved:
|
||||
r.Stored = true
|
||||
r.Answer = "moved: modules." + key + " now names manifest " + h.Manifest + " (" + words + ")"
|
||||
default:
|
||||
r.Held = true
|
||||
r.Answer = "held at manifest " + h.Kept + ": " + words + "; the engines asked again answer in the background, and the hold moves once each states " + strings.Join(h.Needs, ", ")
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// ListHolds is every hold the state holds, by module, or one module's; a value that does not read is named with why.
|
||||
func ListHolds(keys []string, get func(key string) ([]byte, error), module string) (map[string]any, error) {
|
||||
if module != "" {
|
||||
key, err := meshcatalogclient.ModulesKey(module)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
keys = []string{key}
|
||||
}
|
||||
holds, unread := map[string]Hold{}, map[string]string{}
|
||||
for _, key := range slices.Sorted(slices.Values(keys)) {
|
||||
raw, err := get(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if raw == nil {
|
||||
continue
|
||||
}
|
||||
h, err := DecodeHold(raw)
|
||||
if err != nil {
|
||||
unread[key] = err.Error()
|
||||
continue
|
||||
}
|
||||
h.Next = nil // the entry whole is the modules state's form; its digest and commit say which
|
||||
holds[key] = h
|
||||
}
|
||||
return map[string]any{"holds": holds, "unread": unread, "bound": "each hold standing " + HoldBound.String() +
|
||||
" past its since is the finding " + HoldStandingKind + ", its end a key of the state " + HoldBoundsState}, nil
|
||||
}
|
||||
|
||||
// NewHoldAgain is the takeins value indexer_holds writes to ask one module's hold again (release false) or to move it by
|
||||
// a person's word (release true, which takes why: it is a hand act).
|
||||
func NewHoldAgain(module, why, by string, release bool, at time.Time) (string, []byte, error) {
|
||||
key, data, err := NewTakeIn(module, why, by, at)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
var a TakeInAsk
|
||||
if err := json.Unmarshal(data, &a); err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
if release && a.Why == "" {
|
||||
return "", nil, errors.New("say why the hold is moved by hand (why): it is a hand act")
|
||||
}
|
||||
a.Hold = HoldAskAgain
|
||||
if release {
|
||||
a.Hold = HoldRelease
|
||||
}
|
||||
data, err = json.Marshal(a)
|
||||
return holdPrefix + key, data, err
|
||||
}
|
||||
|
||||
// The asks of a hold a takeins value names (TakeInAsk.Hold), and the prefix of its key.
|
||||
const (
|
||||
HoldAskAgain = "again"
|
||||
HoldRelease = "release"
|
||||
holdPrefix = "hold."
|
||||
)
|
||||
@@ -1,121 +0,0 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// fakeBounds is the state hold-bounds: the end each key was armed with, a standing key keeping its end.
|
||||
type fakeBounds map[string]time.Time
|
||||
|
||||
func (b fakeBounds) Arm(_ context.Context, key string, ends time.Time) error {
|
||||
if _, standing := b[key]; !standing {
|
||||
b[key] = ends
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b fakeBounds) Disarm(_ context.Context, key string) error { delete(b, key); return nil }
|
||||
|
||||
// ADR 0632 §2, ADR 0624 §1, a case of TestAMergedManifestIsKeptOnceByItsDigestAndNamedByTheModule. The store holder's
|
||||
// new manifest, which every node-engine reads to find the store, keeps the entry at the old manifest while one counted
|
||||
// node states only the baseline reads (the event manifest-reads-changed not yet heard of it, and the hold asked again by
|
||||
// a person), and moves once every counted node states form:seat-reach. docker, whose manifest stays the same while its
|
||||
// template newly names the image seat's reach (the needs the forge bridge carries), is held the same way. A hold's bound
|
||||
// is a key of hold-bounds; past it the finding manifest-held names n2, cleared when the hold moves; a move by a person's
|
||||
// word names the node it strands.
|
||||
func storeHoldersNewManifestWaitsForEveryCountedEngine(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
const dir, dockerDir = "modules/distribution", "modules/docker"
|
||||
old := `{"module":"distribution","version":"1.0","listens":[{"name":"registry","port":5000,"from":"mesh"}]}`
|
||||
next := `{"module":"distribution","version":"1.0","claims":[{"name":"container-image-registry","scope":"mesh"}],` +
|
||||
`"listens":[{"name":"registry","port":5000,"from":"mesh"}]}`
|
||||
docker := `{"module":"docker","version":"1.0","resources":[{"id":"daemon","type":"file","path":"/etc/docker/daemon.json","template":"files/daemon.json"}]}`
|
||||
dockerNeeds, _ := catalogue.ModuleNeeds([]byte(docker), map[string]string{"files/daemon.json": `"${seat:container-image-registry:reach}"`})
|
||||
if needs, _ := catalogue.ManifestNeeds([]byte(next)); !slices.Contains(needs, "form:seat-reach") || !slices.Contains(dockerNeeds, "form:seat-reach") {
|
||||
t.Fatalf("the new manifest needs %v and docker's %v, want form:seat-reach in both (mesh-sdk ecdbf1a, ModuleNeeds)", needs, dockerNeeds)
|
||||
}
|
||||
ix, mods, _ := manifestsIndexer(t)
|
||||
holds, findings, bounds := newMem(), newMem(), fakeBounds{}
|
||||
stated := map[string][]string{"n1": catalogue.ManifestReads(), "n2": nil} // n2: an engine before the statement, the baseline
|
||||
var asked []string
|
||||
clock := time.Date(2026, 11, 4, 21, 0, 0, 0, time.UTC)
|
||||
ix.Holds, ix.Findings, ix.HoldBounds, ix.Now = holds, findings, bounds, func() time.Time { return clock }
|
||||
ix.Counted = func() (map[string][]string, bool) { return stated, true }
|
||||
ix.AskEngines = func(nodes []string, _ string) { asked = append(asked, nodes...) }
|
||||
heldKey := mcc.Key(mcc.ScopeModule, findingID("hold", "distribution"), HoldStandingKind)
|
||||
|
||||
first := withManifests(mergedMsg(catalogueRepo, map[string]string{dir: "tree1", dockerDir: "d1"}), map[string]string{dir: old, dockerDir: docker}, nil)
|
||||
if err := ix.Apply(ctx, first); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
kept, _ := moduleOf(t, mods, "distribution")
|
||||
keptDocker, _ := moduleOf(t, mods, "docker")
|
||||
if kept.Manifest != meshcatalogclient.ManifestDigest([]byte(old)) || len(holds.vals) != 0 {
|
||||
t.Fatalf("after the first merge modules.distribution names %s with %d holds, want the old manifest and none", kept.Manifest, len(holds.vals))
|
||||
}
|
||||
second := withManifests(mergedMsg(catalogueRepo, map[string]string{dir: "tree2", dockerDir: "d2"}), map[string]string{dir: next, dockerDir: docker}, nil)
|
||||
second.Needs = map[string][]string{dockerDir: dockerNeeds}
|
||||
second.Merge = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
||||
if err := ix.Apply(ctx, second); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stays := func(when string) {
|
||||
t.Helper()
|
||||
m, _ := moduleOf(t, mods, "distribution")
|
||||
if m.Manifest != kept.Manifest || m.Fp2 != kept.Fp2 || m.Commit != kept.Commit || m.Inputs2 == nil || m.Inputs2.Tree != "tree1" {
|
||||
t.Fatalf("%s modules.distribution = manifest %s fp2 %q commit %s, want the old entry kept whole (%s, %q, %s)",
|
||||
when, m.Manifest, m.Fp2, m.Commit, kept.Manifest, kept.Fp2, kept.Commit)
|
||||
}
|
||||
h, err := DecodeHold(holds.vals["distribution"])
|
||||
if err != nil || h.Manifest != meshcatalogclient.ManifestDigest([]byte(next)) || h.Kept != kept.Manifest || !slices.Equal(h.Lacking["n2"], []string{"form:seat-reach"}) || len(h.Lacking) != 1 {
|
||||
t.Fatalf("%s holds.distribution = %+v (%v), want the new manifest held, n2 alone lacking form:seat-reach", when, h, err)
|
||||
}
|
||||
}
|
||||
stays("with n2 stating only the baseline,")
|
||||
if d, _ := moduleOf(t, mods, "docker"); d.Inputs2 == nil || d.Inputs2.Tree != keptDocker.Inputs2.Tree || holds.vals["docker"] == nil {
|
||||
t.Fatalf("docker's template naming the seat's reach: modules.docker at tree %v, holds %v; want its old entry kept and a hold", d.Inputs2, holds.vals["docker"])
|
||||
}
|
||||
if r := ix.HoldAgain(ctx, "distribution", false, "op@shanks"); !r.Held || r.Stored || !slices.Equal(asked, []string{"n2"}) {
|
||||
t.Fatalf("asked again: %+v, engines asked %v; want it held and n2's engine asked again", r, asked)
|
||||
}
|
||||
stays("asked again with n2 still stating only the baseline,")
|
||||
if len(findings.vals) != 0 || !bounds["distribution"].Equal(clock.Add(HoldBound)) {
|
||||
t.Fatalf("within its bound: findings %v, bounds %v; want none and hold-bounds.distribution ending at %v", findings.vals, bounds, clock.Add(HoldBound))
|
||||
}
|
||||
clock = clock.Add(HoldBound)
|
||||
if err := ix.MoveHolds(ctx); err != nil { // what the key's expiry asks (HoldBoundEnded, in the background)
|
||||
t.Fatal(err)
|
||||
}
|
||||
stays("past its bound,")
|
||||
f, err := mcc.DecodeFinding(findings.vals[heldKey])
|
||||
if err != nil || f.Kind != HoldStandingKind || !strings.Contains(f.Summary, "n2 lacks form:seat-reach") {
|
||||
t.Fatalf("past its bound the finding %s = %+v (%v), want manifest-held naming n2", heldKey, f, err)
|
||||
}
|
||||
|
||||
if r := ix.HoldAgain(ctx, "docker", true, "op@shanks"); !r.Stored || !strings.Contains(r.Answer, "strands n2") {
|
||||
t.Fatalf("docker moved by hand: %+v, want it moved and its answer naming n2 stranded", r)
|
||||
}
|
||||
stated["n2"] = catalogue.ManifestReads() // n2's engine states the reading (manifest-reads-changed)
|
||||
if err := ix.MoveHolds(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, _ := moduleOf(t, mods, "distribution")
|
||||
if m.Manifest != meshcatalogclient.ManifestDigest([]byte(next)) || m.Commit != second.Merge || m.Inputs2 == nil || m.Inputs2.Tree != "tree2" || len(holds.vals) != 0 || findings.vals[heldKey] != nil || len(bounds) != 0 {
|
||||
t.Fatalf("once every counted engine states form:seat-reach modules.distribution = manifest %s commit %s, %d holds, finding %s, bounds %v; want the new entry and none",
|
||||
m.Manifest, m.Commit, len(holds.vals), findings.vals[heldKey], bounds)
|
||||
}
|
||||
// Issue 1966: the hold judges as the assigner does (graph.MissingReads): a bootstrapped seat is read by every engine,
|
||||
// so one whose statement names no seat lacks none of it (ADR 0636 §2).
|
||||
stated = map[string][]string{"n1": {"word:module"}}
|
||||
if l, why := ix.lacking([]string{"seat:mesh-catalog", "form:seat-reach"}); why != "" || len(l) != 1 || !slices.Equal(l["n1"], []string{"form:seat-reach"}) {
|
||||
t.Fatalf("an engine stating no seat lacks %v (%s), want form:seat-reach alone", l, why)
|
||||
}
|
||||
}
|
||||
@@ -48,7 +48,6 @@ func TestAMergedManifestIsKeptOnceByItsDigestAndNamedByTheModule(t *testing.T) {
|
||||
t.Run("kept by its digest and named by the module", mergedManifestIsKeptByItsDigestAndNamedByTheModule)
|
||||
t.Run("the same manifest applied again is written once", sameManifestAppliedAgainIsWrittenOnce)
|
||||
t.Run("the seat mesh-catalog's verbs answer from the states", catalogVerbsAnswerFromTheStates)
|
||||
t.Run("a manifest every node reads waits for every counted engine", storeHoldersNewManifestWaitsForEveryCountedEngine) // ADR 0632 §2
|
||||
}
|
||||
|
||||
func mergedManifestIsKeptByItsDigestAndNamedByTheModule(t *testing.T) { // 1449/AC1
|
||||
|
||||
@@ -125,15 +125,6 @@ func (ix *Indexer) applyMoving(ctx context.Context, f forgemsg.Forge, drop bool,
|
||||
next.None2 = in.None2 + ": the forge could not say them: " + namedErr[dir]
|
||||
}
|
||||
}
|
||||
// A manifest every node reads, needing what some counted node-engine does not state, waits in the holds state; the
|
||||
// module keeps its entry until each states it (holds.go, ADR 0632 §2).
|
||||
waits, err := ix.holdBack(ctx, key, next, cur, has)
|
||||
if err != nil {
|
||||
return w, err
|
||||
}
|
||||
if waits {
|
||||
continue
|
||||
}
|
||||
overlay[key] = next
|
||||
}
|
||||
// A module of this repository the merge commit holds no module.json for is gone from the trunk, and so is its key
|
||||
|
||||
@@ -908,7 +908,7 @@ func (ix *Indexer) dry() *dryRun {
|
||||
// requirements, findings, seat definitions, holds and their bounds are not overlaid, nothing is said, and no hold is
|
||||
// weighed (Holds and Counted nil hold nothing); ReadFile only reads.
|
||||
Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil,
|
||||
Holds: nil, Counted: nil, AskEngines: nil, HoldBounds: nil, OwnBuildState: nil, underSeats: true,
|
||||
OwnBuildState: nil, underSeats: true,
|
||||
// A dry run keeps no catalog graph revision (catalog_revisions.go): its revision is the take-in's to keep.
|
||||
Revisions: nil, SayCatalogRevision: nil, SayFill: nil}
|
||||
x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) }
|
||||
|
||||
@@ -1127,7 +1127,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged",
|
||||
"JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk",
|
||||
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn",
|
||||
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Holds", "Counted", "AskEngines", "HoldBounds", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill"} {
|
||||
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill"} {
|
||||
named[f] = true
|
||||
}
|
||||
ty := reflect.TypeOf(Indexer{})
|
||||
|
||||
@@ -43,9 +43,6 @@ type TakeInAsk struct {
|
||||
Say bool `json:"say,omitempty"`
|
||||
// SeatDefinitions asks a contracts module's seat definitions to be read again now (indexer_seat_definitions_again,
|
||||
// DLV-I19): its trunk's, or with Pull and Head a pull request's head.
|
||||
// Hold asks the module's mesh-wide hold (holds.go): HoldAskAgain weighs it again now, HoldRelease moves it by a
|
||||
// person's word (indexer_holds).
|
||||
Hold string `json:"hold,omitempty"`
|
||||
SeatDefinitions bool `json:"seat-definitions,omitempty"`
|
||||
Pull int `json:"pull,omitempty"`
|
||||
Head string `json:"head,omitempty"`
|
||||
|
||||
@@ -23,7 +23,6 @@ var FirstViewTools = []ViewedTool{
|
||||
// ReadTools are the indexer's ordinary read tools, one per state its readers read (design 25 § 12.2).
|
||||
var ReadTools = []ViewedTool{
|
||||
{Tool: "indexer_read_manifests", State: "manifests", Words: "every manifest the indexer keeps, by its digest."},
|
||||
{Tool: "indexer_read_modules", State: "modules", Words: "every module the indexer holds, its fingerprint, inputs and manifest digest."},
|
||||
{Tool: "indexer_read_predefined", State: PredefinedState, Words: "the predefined config values of each module on the trunk."},
|
||||
{Tool: "indexer_read_config_candidates", State: ConfigCandidatesState, Words: "the config declarations of each open pull request."},
|
||||
{Tool: "indexer_read_candidates", State: "candidates", Words: "the candidate groups: the pull requests sharing a branch, their heads and the modules each touches."},
|
||||
|
||||
@@ -121,7 +121,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {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: index.PackagesState, History: index.PackagesHistory},
|
||||
{Name: views.MarksState, History: 1}, {Name: mcc.FindingsState, History: 5}, {Name: views.StatusState, History: 1},
|
||||
{Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5},
|
||||
{Name: index.HoldsState, History: 5}, {Name: index.EngineReadsState, History: 1}, {Name: index.HoldBoundsState, History: 1, KeyTTL: true}, {Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17, ADR 0632 §2
|
||||
{Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
// Milestone 47/G W4.2 (design 47, Two graphs; ADR 0619 §2): the catalog graph revisions are objects in the old section,
|
||||
@@ -157,7 +157,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
// The registry seats' published and the builder's numbered fill each package's series (DLV-I5, design 51 *The versions*).
|
||||
if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-assigner.views.assignments",
|
||||
"mesh-delivery-builder.version.numbered", "mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**",
|
||||
"mesh-delivery.source-moved", "node-engine.manifest-reads-changed", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) { // ADR 0632 §2: the counted engines' reads
|
||||
"mesh-delivery.source-moved", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
t.Errorf("consumes = %v, want %v", m.Consumes, want)
|
||||
}
|
||||
if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
|
||||
@@ -169,7 +169,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
// Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths:
|
||||
// ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a
|
||||
// commit is in another (issue 1900).
|
||||
if want := []string{"mesh-delivery-assigner.assignments", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:node-engine.manifest-reads", "seat:mesh-delivery.source", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
if want := []string{"mesh-delivery-assigner.assignments", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
// 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle
|
||||
@@ -188,13 +188,27 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
}
|
||||
want = append(want, "indexer_first_views", "indexer_first_view_again", index.ConfigReadersTool, index.BusStateTool, index.RequirementsTool, index.PackageUsersTool, index.SayAcceptedTool, // DLV-F3r, DLV-I10, DLV-I11, DLV-I7, DLV-I8, DLV-I9c
|
||||
index.SeatDefinitionTool, index.SeatDefinitionsAgainTool, index.OutOfStepTool, // DLV-I19
|
||||
index.HoldsTool, index.EngineReadsTool, // ADR 0632 §2: the mesh-wide holds and the engines' reads they weigh
|
||||
index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TryCandidatesTool, index.DropTryCandidateTool, index.RecoverTriesTool, index.TakeInMergeTool, index.FingerprintsAgainTool, // issue 1966; 1585/T12, T13 (ADR 0633 §5)
|
||||
// milestone 47/G W4.12 (design 47, the indexer's tools): the catalog graph revisions read and steered.
|
||||
index.CatalogRevisionTool, index.CatalogRevisionsTool, index.CandidateGraphsTool, index.RecomputeTool, index.DropCandidateGraphTool)
|
||||
if !reflect.DeepEqual(m.Tools, want) || len(want) != 44 || slices.Contains(m.Tools, "modules") {
|
||||
if !reflect.DeepEqual(m.Tools, want) || len(want) != 41 || slices.Contains(m.Tools, "modules") {
|
||||
t.Errorf("tools = %v, want %v", m.Tools, want)
|
||||
}
|
||||
// Design 47 G §196, §185 (test 61): no state modules, no tool or view of it, and no mesh-wide hold with its states, tools
|
||||
// and the engines' reads it weighed.
|
||||
for _, s := range m.State {
|
||||
if slices.Contains([]string{"modules", "holds", "engine-reads", "hold-bounds"}, s.Name) {
|
||||
t.Errorf("the indexer declares the state %s, which design 47 G retires", s.Name)
|
||||
}
|
||||
}
|
||||
for _, retired := range []string{"indexer_read_modules", "indexer_holds", "indexer_engine_reads"} {
|
||||
if slices.Contains(m.Tools, retired) {
|
||||
t.Errorf("the indexer serves %s, which design 47 G retires", retired)
|
||||
}
|
||||
}
|
||||
if slices.Contains(m.Emits, "views.modules") || slices.Contains(m.Consumes, "node-engine.manifest-reads-changed") {
|
||||
t.Errorf("the indexer says views.modules or hears manifest-reads-changed: emits %v, consumes %v", m.Emits, m.Consumes)
|
||||
}
|
||||
// Every read tool of the indexer a reader of this repository invokes is one it serves (the checker's seventh first
|
||||
// view, indexer_read_seat_definitions, was invoked and not served).
|
||||
readers, _ := filepath.Glob("../*/module.json")
|
||||
@@ -212,7 +226,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
}
|
||||
for _, inv := range r.Invokes {
|
||||
// The grants part (W8) and the assigner read modules until their own subtasks move them to revision; the join removes this line.
|
||||
if (p == "../mesh-grants/module.json" || p == "../mesh-delivery-assigner/module.json") && inv == "mesh-delivery-indexer.modules" {
|
||||
if (p == "../mesh-grants/module.json" || p == "../mesh-delivery-assigner/module.json") && inv == "mesh-delivery-indexer.modules" ||
|
||||
inv == "mesh-delivery-indexer.indexer_read_modules" { // design 47 G §196: the readers' own subtasks move them to the catalog half
|
||||
continue
|
||||
}
|
||||
if tool, ok := strings.CutPrefix(inv, "mesh-delivery-indexer."); ok && !slices.Contains(m.Tools, tool) {
|
||||
@@ -264,7 +279,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if len(m.Streams) != 0 {
|
||||
t.Errorf("streams = %+v, want none", m.Streams)
|
||||
}
|
||||
for _, state := range []string{"candidates", "config-candidates", "findings", "graph", "heard", "manifests", "modules", "packages", "predefined", "seat-definitions", "taken", "verdicts"} {
|
||||
for _, state := range []string{"candidates", "config-candidates", "findings", "graph", "heard", "manifests", "packages", "predefined", "seat-definitions", "taken", "verdicts"} {
|
||||
if !slices.Contains(m.Emits, views.EventPrefix+state) {
|
||||
t.Errorf("emits = %v: want %s%s", m.Emits, views.EventPrefix, state)
|
||||
}
|
||||
|
||||
@@ -26,7 +26,6 @@
|
||||
"verdicts",
|
||||
"graph",
|
||||
"indexer_read_manifests",
|
||||
"indexer_read_modules",
|
||||
"indexer_read_predefined",
|
||||
"indexer_read_config_candidates",
|
||||
"indexer_read_candidates",
|
||||
@@ -47,8 +46,6 @@
|
||||
"indexer_seat_definition",
|
||||
"indexer_seat_definitions_again",
|
||||
"indexer_out_of_step",
|
||||
"indexer_holds",
|
||||
"indexer_engine_reads",
|
||||
"indexer_pending",
|
||||
"indexer_take_in_pending",
|
||||
"indexer_drop_pending",
|
||||
@@ -75,7 +72,6 @@
|
||||
"mesh-delivery-checker.verdict",
|
||||
"mesh-delivery-forge-bridge.forge.**",
|
||||
"mesh-delivery.source-moved",
|
||||
"node-engine.manifest-reads-changed",
|
||||
"npm-package-registry.published"
|
||||
],
|
||||
"state": [
|
||||
@@ -96,9 +92,6 @@
|
||||
{ "name": "first-views", "history": 1 },
|
||||
{ "name": "requirements", "history": 5 },
|
||||
{ "name": "seat-definitions", "history": 5 },
|
||||
{ "name": "holds", "history": 5 },
|
||||
{ "name": "engine-reads", "history": 1 },
|
||||
{ "name": "hold-bounds", "history": 1, "key-ttl": true },
|
||||
{ "name": "pending", "history": 1 },
|
||||
{ "name": "unjudged", "history": 1 }
|
||||
],
|
||||
@@ -114,7 +107,6 @@
|
||||
"views.graph",
|
||||
"views.heard",
|
||||
"views.manifests",
|
||||
"views.modules",
|
||||
"views.packages",
|
||||
"views.predefined",
|
||||
"views.seat-definitions",
|
||||
@@ -136,7 +128,6 @@
|
||||
"mesh-delivery-checker.checker_tries",
|
||||
"mesh-delivery-forge-bridge.bridge_merged_message",
|
||||
"seat:node-engine.composition",
|
||||
"seat:node-engine.manifest-reads",
|
||||
"seat:mesh-delivery.source",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-watchdogs.presence"
|
||||
|
||||
Reference in New Issue
Block a user