Let the indexer write a verdict on every manifest it holds and say each refusal (ADR 0578 §1)
Issue 1585, S9: verdicts keyed by digest, written only when they differ; a module's value carries a refusal; a new refusal is said as manifest-refused. Judged at start (a new judge judges again) and after every merge. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -291,7 +291,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return err
|
||||
}
|
||||
defer calls.Close()
|
||||
// The verdicts on manifests (ADR 0578 §1): a state this delivery brings, bound when it appears.
|
||||
verdicts := process.AwaitBucket(ctx, js, conn, index.VerdictsBucket, "no verdict on a manifest is written", logger.Printf)
|
||||
ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known,
|
||||
Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(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),
|
||||
ReadOpen: openVia(calls.Call)}
|
||||
@@ -385,6 +388,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err := ix.Reconcile(ctx); err != nil {
|
||||
logger.Printf("dropping the modules the trunk does not hold at start: %v", err)
|
||||
}
|
||||
// Every manifest judged at the start, so a judge that changed with this program judges every one again (ADR 0578
|
||||
// §1), once the verdicts state is there; each merge judges them again (index.ApplyMessage).
|
||||
verdicts.Ready(func(jetstream.KeyValue) {
|
||||
j, err := ix.JudgeAll(ctx)
|
||||
if err != nil {
|
||||
logger.Printf("judging the manifests at start: %v", err)
|
||||
return
|
||||
}
|
||||
logger.Printf("judged %d manifests at start: %d verdicts written, %d refused", j.Judged, j.Written, len(j.Refused))
|
||||
})
|
||||
// A build module says each time it wrote its toolchains key: its stream, and the indexer's durable consumer on it, are
|
||||
// made by the controller as the forge's are. Where the stream is not there yet the indexer says so and is started again
|
||||
// with the build module, as with the gate's source.
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// sayRefused says the indexer's event manifest-refused (novox/hq ADR 0578 §1): mesh.mod.<module>.event.manifest-refused,
|
||||
// with the SDK's envelope headers and the digest and judge as the bus's dedup id, so one refusal is said once.
|
||||
func sayRefused(js jetstream.JetStream, cred process.Account, logf func(string, ...any)) func(context.Context, delivery.ManifestRefused) {
|
||||
return func(ctx context.Context, r delivery.ManifestRefused) {
|
||||
data, err := json.Marshal(r)
|
||||
if err != nil {
|
||||
logf("judge: %s for %s is not said: %v", delivery.ManifestRefusedEvent, r.Module, err)
|
||||
return
|
||||
}
|
||||
msg := nats.NewMsg(cred.EventSubject(delivery.ManifestRefusedEvent))
|
||||
id := "manifest-refused-" + r.Digest + "-" + r.Judge
|
||||
msg.Header.Set("x-event-id", id)
|
||||
msg.Header.Set("x-source", cred.Module)
|
||||
msg.Header.Set("x-node", cred.Node)
|
||||
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
|
||||
msg.Header.Set("content-type", "application/json")
|
||||
msg.Data = data
|
||||
pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||
defer cancel()
|
||||
if _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)); err != nil {
|
||||
logf("judge: %s", fmt.Errorf("saying %s for %s: %w", delivery.ManifestRefusedEvent, r.Module, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -52,6 +52,12 @@ type Indexer struct {
|
||||
Predefined, ConfigCandidates Store
|
||||
// Taken is the taken bucket: per repository the last merge taken in (taken.go, ADR 0556); nil writes none.
|
||||
Taken Store
|
||||
// Verdicts is the verdicts state (verdicts.go, novox/hq ADR 0578 §1); nil judges nothing.
|
||||
Verdicts Store
|
||||
// JudgeID names the judge a verdict says made it; empty is judge.Identity, the running program's.
|
||||
JudgeID string
|
||||
// SayRefused says the event manifest-refused for a refusal newly written; nil says nothing.
|
||||
SayRefused func(ctx context.Context, r delivery.ManifestRefused)
|
||||
// Now is the clock of a taken value; nil is time.Now.
|
||||
Now func() time.Time
|
||||
Builds *Builds
|
||||
@@ -131,6 +137,10 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f delivery.Forge) (Applied,
|
||||
if err := ix.configCandidatesAfterMerge(ctx, f, writtenModules(w.written)); err != nil {
|
||||
return applied, err
|
||||
}
|
||||
// Every manifest judged again: a rule between manifests can turn on the one this merge moved (ADR 0578 §1).
|
||||
if _, err := ix.JudgeAll(ctx); err != nil {
|
||||
ix.logf("judge: the manifests are not judged after %s#%d: %v", f.Repository, f.Number, err)
|
||||
}
|
||||
}
|
||||
ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-indexer/judge"
|
||||
)
|
||||
|
||||
// The indexer judges every manifest it holds, once, with its own judge (novox/hq ADR 0578 §1): it writes each verdict in
|
||||
// its state verdicts (delivery.VerdictsBucket), one key per manifest digest, only when the verdict differs from the one
|
||||
// held; a module's value in the modules state carries a refusal (Verdict, Problems), so a reader of modules takes it
|
||||
// without a second read; a new refusal is said as the event manifest-refused. It judges at its start, so a judge that
|
||||
// changed (a new indexer) judges every manifest again, and after every merge it takes in, since a rule between manifests
|
||||
// can turn on any of them. Nothing polls.
|
||||
|
||||
// VerdictsBucket is the indexer's state of verdicts.
|
||||
var VerdictsBucket = delivery.VerdictsBucket
|
||||
|
||||
// Judged is what one pass of the judge did: the manifests judged, the verdicts written, and the modules refused.
|
||||
type Judged struct {
|
||||
Judged, Written int
|
||||
Refused map[string][]string
|
||||
}
|
||||
|
||||
// JudgeAll judges every manifest the modules state names, as one catalogue with the predefined list the indexer keeps,
|
||||
// and writes what differs. A manifest the manifests state does not hold is left unjudged and said; a write that fails is
|
||||
// said and the pass goes on, so one key never costs the others their verdict.
|
||||
func (ix *Indexer) JudgeAll(ctx context.Context) (Judged, error) {
|
||||
var out Judged
|
||||
if ix.Verdicts == nil || ix.Modules == nil || ix.Manifests == nil {
|
||||
return out, nil
|
||||
}
|
||||
held, err := ix.heldModules(ctx)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
raws := map[string][]byte{}
|
||||
digests := map[string]string{}
|
||||
for _, name := range slices.Sorted(maps.Keys(held)) {
|
||||
digest := held[name].Manifest
|
||||
if digest == "" {
|
||||
continue
|
||||
}
|
||||
value, _, found, err := ix.Manifests.Get(ctx, digest)
|
||||
if err != nil {
|
||||
return out, fmt.Errorf("reading manifests.%s of %s: %w", digest, name, err)
|
||||
}
|
||||
if !found {
|
||||
ix.logf("judge: the manifest of %s (digest %s) is not in the manifests state, so it is not judged", name, digest)
|
||||
continue
|
||||
}
|
||||
kept, err := delivery.DecodeManifest(value)
|
||||
if err != nil {
|
||||
ix.logf("judge: manifests.%s of %s cannot be read, so it is not judged: %v", digest, name, err)
|
||||
continue
|
||||
}
|
||||
raws[name] = []byte(kept.JSON)
|
||||
digests[name] = digest
|
||||
}
|
||||
refused, unnamed := judge.Shelf(raws, ix.predefinedList(ctx), 0)
|
||||
for _, p := range unnamed {
|
||||
ix.logf("judge: a problem between the manifests that no one module's removal ends, so it refuses none: %s", p)
|
||||
}
|
||||
out.Refused = refused
|
||||
id := ix.JudgeID
|
||||
if id == "" {
|
||||
id = judge.Identity()
|
||||
}
|
||||
for _, name := range slices.Sorted(maps.Keys(digests)) {
|
||||
out.Judged++
|
||||
v := delivery.ManifestVerdict{Module: name, State: delivery.VerdictAccepted, Judge: id, At: ix.now()}
|
||||
if problems := refused[name]; len(problems) > 0 {
|
||||
v.State, v.Problems = delivery.VerdictRefused, problems
|
||||
}
|
||||
wrote, err := ix.writeVerdict(ctx, digests[name], v)
|
||||
if errors.Is(err, process.ErrNotYet) {
|
||||
return out, fmt.Errorf("the state %s is not on the bus yet: %w", VerdictsBucket, err)
|
||||
}
|
||||
if err != nil {
|
||||
ix.logf("judge: the verdict on %s (digest %s) is not written: %v", name, digests[name], err)
|
||||
continue
|
||||
}
|
||||
if wrote {
|
||||
out.Written++
|
||||
if v.Refused() {
|
||||
ix.logf("judge: %s", v.Why())
|
||||
if ix.SayRefused != nil {
|
||||
ix.SayRefused(ctx, delivery.RefusedOf(digests[name], v))
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := ix.carryVerdict(ctx, name, digests[name], v); err != nil {
|
||||
ix.logf("judge: modules.%s does not carry its verdict: %v", name, err)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// writeVerdict writes v under the digest's key unless the verdict held there says the same (ManifestVerdict.Same); wrote
|
||||
// says it was written.
|
||||
func (ix *Indexer) writeVerdict(ctx context.Context, digest string, v delivery.ManifestVerdict) (wrote bool, err error) {
|
||||
key, err := delivery.VerdictsKey(digest)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
err = write(ctx, ix.Verdicts, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
wrote = false
|
||||
if found {
|
||||
if cur, err := delivery.DecodeManifestVerdict(value); err == nil && cur.Same(v) {
|
||||
return nil, false, nil
|
||||
}
|
||||
}
|
||||
wrote = true
|
||||
next, err := delivery.EncodeManifestVerdict(v)
|
||||
return next, false, err
|
||||
})
|
||||
return wrote, err
|
||||
}
|
||||
|
||||
// carryVerdict puts a refusal into the module's value, and takes one out once the manifest is accepted, while the value
|
||||
// still names the digest judged; an accepted module's value carries nothing, so judging writes no value of a module the
|
||||
// judge accepts (a reader takes a value without a verdict as accepted, delivery.TakeManifestVerdict).
|
||||
func (ix *Indexer) carryVerdict(ctx context.Context, name, digest string, v delivery.ManifestVerdict) error {
|
||||
key, err := delivery.ModulesKey(name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
if !found {
|
||||
return nil, false, nil
|
||||
}
|
||||
cur, err := delivery.DecodeModule(value)
|
||||
if err != nil || cur.Manifest != digest {
|
||||
return nil, false, nil // gone, unreadable, or moved to another manifest since: the next pass judges it
|
||||
}
|
||||
state := ""
|
||||
if v.Refused() {
|
||||
state = delivery.VerdictRefused
|
||||
}
|
||||
if cur.Verdict == state && slices.Equal(cur.Problems, v.Problems) {
|
||||
return nil, false, nil
|
||||
}
|
||||
cur.Verdict, cur.Problems = state, v.Problems
|
||||
if state == "" {
|
||||
cur.Problems = nil
|
||||
}
|
||||
next, err := delivery.EncodeModule(cur)
|
||||
return next, false, err
|
||||
})
|
||||
}
|
||||
|
||||
// predefinedList is the catalogue's predefined list the indexer keeps (writePredefined), or nil when it keeps none or
|
||||
// it cannot be read: the judge then takes each fallback as its value declares it.
|
||||
func (ix *Indexer) predefinedList(ctx context.Context) catalogue.PredefinedList {
|
||||
if ix.Predefined == nil {
|
||||
return nil
|
||||
}
|
||||
value, _, found, err := ix.Predefined.Get(ctx, delivery.PredefinedListKey)
|
||||
if err != nil || !found {
|
||||
return nil
|
||||
}
|
||||
var kept delivery.PredefinedKept
|
||||
if json.Unmarshal(value, &kept) != nil {
|
||||
return nil
|
||||
}
|
||||
list, err := catalogue.ParsePredefined(kept.List)
|
||||
if err != nil {
|
||||
ix.logf("judge: the predefined list kept cannot be read, so fallbacks are judged as their values declare them: %v", err)
|
||||
return nil
|
||||
}
|
||||
return list
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The indexer judges every manifest it holds and writes each verdict only when it differs; a module's value carries a
|
||||
// refusal, which is said once as manifest-refused; a new judge writes again only what it judges otherwise; a manifest
|
||||
// mended clears the module's refusal (novox/hq ADR 0578 §1).
|
||||
func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
ix := &Indexer{Modules: newMem(), Manifests: newMem(), Verdicts: newMem(), JudgeID: "judge-1",
|
||||
Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
|
||||
var said []delivery.ManifestRefused
|
||||
ix.SayRefused = func(_ context.Context, r delivery.ManifestRefused) { said = append(said, r) }
|
||||
hold := func(module, raw string) string {
|
||||
digest, err := ix.keepManifest(ctx, module, raw)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := delivery.EncodeModule(delivery.Module{FP2: "src2:" + strings.Repeat("a", 64), Manifest: digest,
|
||||
Repository: "novox/mesh-catalog", Commit: "c"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ix.Modules.(*memStore).put(module, value)
|
||||
return digest
|
||||
}
|
||||
good := hold("good", `{"module":"good","version":"1"}`)
|
||||
bad := hold("bad", `{"module":"bad","version":"1","requires":["amqp"]}`)
|
||||
j, err := ix.JudgeAll(ctx)
|
||||
if err != nil || j.Judged != 2 || j.Written != 2 || len(j.Refused) != 1 {
|
||||
t.Fatalf("the first pass: %+v %v", j, err)
|
||||
}
|
||||
verdict := func(digest string) delivery.ManifestVerdict {
|
||||
value, _, found, _ := ix.Verdicts.Get(ctx, digest)
|
||||
v, err := delivery.DecodeManifestVerdict(value)
|
||||
if !found || err != nil {
|
||||
t.Fatalf("no verdict under %s: %v", digest, err)
|
||||
}
|
||||
return v
|
||||
}
|
||||
if v := verdict(good); v.Refused() || v.Judge != "judge-1" {
|
||||
t.Fatalf("good: %+v", v)
|
||||
}
|
||||
if v := verdict(bad); !v.Refused() || !strings.Contains(v.Problems[0], `requires "amqp"`) {
|
||||
t.Fatalf("bad: %+v", v)
|
||||
}
|
||||
module := func(name string) delivery.Module {
|
||||
value, _, _, _ := ix.Modules.Get(ctx, name)
|
||||
m, err := delivery.DecodeModule(value)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return m
|
||||
}
|
||||
if m := module("bad"); m.Verdict != delivery.VerdictRefused || len(m.Problems) != 1 {
|
||||
t.Fatalf("modules.bad carries %+v", m)
|
||||
}
|
||||
if m := module("good"); m.Verdict != "" {
|
||||
t.Fatalf("modules.good carries %q", m.Verdict)
|
||||
}
|
||||
if len(said) != 1 || said[0].Module != "bad" || said[0].Digest != bad || said[0].Judge != "judge-1" {
|
||||
t.Fatalf("said %+v", said)
|
||||
}
|
||||
// A new judge that judges alike writes nothing and says nothing again.
|
||||
ix.JudgeID = "judge-2"
|
||||
if j, err := ix.JudgeAll(ctx); err != nil || j.Written != 0 || len(said) != 1 {
|
||||
t.Fatalf("a judge that judges alike: %+v %v, said %d", j, err, len(said))
|
||||
}
|
||||
// The manifest mended: its new digest is accepted and the module's refusal goes.
|
||||
hold("bad", `{"module":"bad","version":"2"}`)
|
||||
if j, err := ix.JudgeAll(ctx); err != nil || j.Written != 1 || len(j.Refused) != 0 {
|
||||
t.Fatalf("mended: %+v %v", j, err)
|
||||
}
|
||||
if m := module("bad"); m.Verdict != "" || m.Problems != nil {
|
||||
t.Fatalf("modules.bad still carries %+v", m)
|
||||
}
|
||||
// Without the verdicts state the indexer judges nothing.
|
||||
ix.Verdicts = nil
|
||||
if j, err := ix.JudgeAll(ctx); err != nil || j.Judged != 0 {
|
||||
t.Fatalf("without the state: %+v %v", j, err)
|
||||
}
|
||||
}
|
||||
@@ -81,12 +81,13 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if m.Module != "mesh-delivery-indexer" {
|
||||
t.Errorf("module = %q", m.Module)
|
||||
}
|
||||
// 1126: placement is the seat mesh-placement's (ADR 0567): the indexer holds no placement state and says no event.
|
||||
if len(m.Emits) != 0 {
|
||||
t.Errorf("emits = %v, want none", m.Emits)
|
||||
// 1126: placement is the seat mesh-placement's (ADR 0567): the indexer holds no placement state. ADR 0578 §1: it says
|
||||
// a refused manifest as manifest-refused, its one event.
|
||||
if !reflect.DeepEqual(m.Emits, []string{"manifest-refused"}) {
|
||||
t.Errorf("emits = %v, want manifest-refused alone", 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}}; !reflect.DeepEqual(m.State, want) {
|
||||
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}}; !reflect.DeepEqual(m.State, want) {
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
|
||||
@@ -19,7 +19,11 @@
|
||||
{ "name": "predefined", "history": 5 },
|
||||
{ "name": "config-candidates", "history": 5 },
|
||||
{ "name": "taken", "history": 1 },
|
||||
{ "name": "takeins", "history": 1 }
|
||||
{ "name": "takeins", "history": 1 },
|
||||
{ "name": "verdicts", "history": 5 }
|
||||
],
|
||||
"emits": [
|
||||
"manifest-refused"
|
||||
],
|
||||
"reads": [
|
||||
"mesh-build-go.toolchains",
|
||||
|
||||
Reference in New Issue
Block a user