Merge branch 'feat/47-G-indexer-dev2' into feat/47-G-dev2-join

This commit is contained in:
2026-11-11 21:06:21 +01:00
39 changed files with 2616 additions and 119 deletions
+11 -2
View File
@@ -680,13 +680,22 @@ type ActingRule struct {
Step string `json:"step"`
}
// ArtifactPackage is the package's name read from the revision's own files at the artifact's path (the module of its go.mod, the name of its package.json, the manifest's image name), written at take-in (design 47 G section 676).
type ArtifactPackage struct {
// go, npm or image
Ecosystem string `json:"ecosystem"`
Name string `json:"name"`
}
// Artifact is one artifact of a module in the catalog graph entry.
type Artifact struct {
// the artifact's kind
Kind string `json:"kind,omitempty"`
// the artifact's language as its manifest says it (go, typescript, ...); a step's build seat is read from it (design 47 G section 323)
Language string `json:"language"`
Ref ArtifactRef `json:"ref"`
Language string `json:"language"`
// the package's name read from the revision's own files at the artifact's path (the module of its go.mod, the name of its package.json, the manifest's image name), written at take-in (design 47 G section 676)
Package *ArtifactPackage `json:"package,omitempty"`
Ref ArtifactRef `json:"ref"`
}
// ArtifactRef is an artifact reference, <module>:<artifact>, such as mesh-grants:program (design 47 *Dependencies between artifacts*).
+17
View File
@@ -2711,6 +2711,23 @@
"kind": {
"type": "string",
"description": "the artifact's kind"
},
"package": {
"type": "object",
"description": "the package's name read from the revision's own files at the artifact's path (the module of its go.mod, the name of its package.json, the manifest's image name), written at take-in (design 47 G section 676)",
"properties": {
"ecosystem": {
"type": "string",
"description": "go, npm or image"
},
"name": {
"type": "string"
}
},
"required": [
"ecosystem",
"name"
]
}
},
"required": [
+9
View File
@@ -191,12 +191,21 @@ export interface ActingRule {
"step": string;
}
/** the package's name read from the revision's own files at the artifact's path (the module of its go.mod, the name of its package.json, the manifest's image name), written at take-in (design 47 G section 676) */
export interface ArtifactPackage {
/** go, npm or image */
"ecosystem": string;
"name": string;
}
/** one artifact of a module in the catalog graph entry */
export interface Artifact {
/** the artifact's kind */
"kind"?: string;
/** the artifact's language as its manifest says it (go, typescript, ...); a step's build seat is read from it (design 47 G section 323) */
"language": string;
/** the package's name read from the revision's own files at the artifact's path (the module of its go.mod, the name of its package.json, the manifest's image name), written at take-in (design 47 G section 676) */
"package"?: ArtifactPackage;
"ref": ArtifactRef;
}
@@ -288,11 +288,15 @@ func main() {
tools = append(tools, catalogTools(trunk.State)...)
// The seat mesh-catalog's verb revision, a first view of the catalog half answered in this runtime of its own to exactly
// the parts whose entry declares it (novox/hq ADR 0663).
tools = append(tools, index.RevisionTool(revisions, index.Module))
tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}))
tools = append(tools, index.CatalogRevisionTools(revisions, asks, time.Now, func(what, why string) error {
_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.DropCandidateGraphTool, Verb: "drop-candidate-graph"})
return err
})...)
tools = append(tools, index.RegistrationTools(keptRead{stdio.State(index.ProposalsState)}, asks, time.Now, func(cause, verb, what, why string) error {
_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause, Verb: verb})
return err
})...)
tools = append(tools, viewTools(os.Getenv, trunk.State, time.Now)...)
tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, keptRead{stdio.State(index.OwnBuildStateName)}, asks, time.Now)...)
tools = append(tools, index.FingerprintsTools(asks, time.Now, func(what, why string) error {
@@ -625,7 +625,13 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
t.Fatal(err)
}
acceptedHeard.Store(0)
waiter, err := ctl.Subscribe("mesh.seat.mesh-catalog.event.manifest-accepted", func(*nats.Msg) { acceptedHeard.Add(1) })
waiter, err := ctl.Subscribe("mesh.seat.mesh-catalog.event.manifest-accepted", func(m *nats.Msg) {
// manifest-accepted is said for each manifest the indexer accepts (design 47 G §4 of the seat's events): only a body that
// is a views event, a wake by the views family, counts.
if _, err := views.DecodeEvent(m.Data); err == nil {
acceptedHeard.Add(1)
}
})
if err != nil {
t.Fatal(err)
}
@@ -420,7 +420,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.PackagesBucket}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call),
OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, TryRecovery: awaitedStore{b: tryRecovery}, Graph: graph,
SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats,
Message: messageVia(calls.CallKept), Revisions: revisions, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1)
Message: messageVia(calls.CallKept), Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.EntryIndexBucket, "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1)
// Each catalog revision may bring another holder of a build seat: the build modules are found again (buildmodules.go).
SayCatalogRevision: refreshingBuildModules(sayCatalogRevision(conn, cred), &buildFollow),
SayFill: sayFill(conn, cred),
@@ -428,6 +428,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// until then every merge is taken in at once, as before.
Pending: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)},
SayTakeIn: sayTakeIn(js, cred)}
// The registrations and the changes that wait for their warrant (design 47 G §2, §667): their states are bound when they
// appear; a warrant is heard on the router's decided event for this module, below.
registry := newRegistry(js, conn, cred, calls.Call,
awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.RegistrationsBucket, "no registration is kept or written", logger.Printf)},
awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.ProposalsBucket, "no change of a registration waits for its warrant", logger.Printf)},
routerRecordUnread, logger.Printf)
ix.Registry = registry
ix.SeedModules = func(context.Context) ([]index.SeedModule, error) {
return nil, errors.New("the first registrations are not composed by this build: what the build seats answer steps with is not read yet (design question)")
}
// Which build this program is, fixed here once, before it hears any merge and never asked again (index.FixOwnBuild): the
// node-engine names the successor while it replaces this process, and an old process that learnt it later would take
// revisions in with its old judge.
@@ -623,7 +633,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
checksCons = c
return nil
})
// The versions each package's series holds (DLV-I5): the registry seats' published and the builder's numbered, on the
// The versions each package's series holds (DLV-I5): the registry seats' published, 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(meshdelivery.SubjectSourceMovedAll, func(m jetstream.Msg) { source.heard(ctx, m) }); err != nil {
@@ -634,15 +644,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
_ = m.Ack() // nothing else the indexer consumes rides the shared stream
}
}
if err := hub.Handle(decidedSubject(cred.Module), func(m jetstream.Msg) { heardDecided(ctx, registry, m, logger.Printf) }); err != nil {
return err
}
for subject := range publishedSeats {
if err := hub.Handle(subject, versions); err != nil {
return err
}
}
if err := hub.Handle(numberedSubject, versions); err != nil {
return err
// A change decided while the indexer was down is applied or ended now (§667, A restart).
if err := registry.Restart(ctx); err != nil {
logger.Printf("the open changes of the registrations are not all read again at the start: %v", err)
}
_ = notHeard.Bind(process.EventsStream, "no package's series moves, no node's manifest-reads-changed and no assignments view is heard until the indexer starts again", func() error {
_ = notHeard.Bind(process.EventsStream, "no package's series moves, no warrant for a change of a registration is heard, no node's manifest-reads-changed and no assignments view is heard until the indexer starts again", func() error {
if err := hub.Start(ctx, js, cred); err != nil {
stopped = cmp.Or(stopped, err)
return err
@@ -0,0 +1,136 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
gitseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/git"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/process"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"mesh-delivery-indexer/internal/index"
)
// The registrations' parts as the indexer's program wires them (design 47 G §2 to §4, §667).
// decidedSubject is where the router says each warrant for this module (the seat operator-channel's event decided).
func decidedSubject(module string) string { return oc.SubjectDecided(module) }
// publishAs publishes one message on subject under a duplicate id, with the headers every event of the indexer carries.
func publishAs(conn *nats.Conn, cred process.Account, subject, id string, body any) error {
data, err := json.Marshal(body)
if err != nil {
return err
}
msg := nats.NewMsg(subject)
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.Header.Set(nats.MsgIdHdr, id)
msg.Data = data
return conn.PublishMsg(msg)
}
// historySubject is where one module's versions land on the stream registration-history.
func historySubject(module string) string {
return process.ModuleStreamSubject(index.Module, index.RegistrationHistoryEvent+"."+module)
}
// newRegistry is the Registry over the indexer's own states, the stream registration-history, the seat git, and the seat
// operator-channel's accepts. where is the read of the router's own record of an ask (design question D-37, open).
func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account, call toolCaller, registrations, proposals index.Store, where func(id string) (string, *oc.Warrant, error), logf func(string, ...any)) *index.Registry {
return &index.Registry{Registrations: registrations, Proposals: proposals,
History: func(ctx context.Context, r meshcatalog.Registration) error {
data, err := json.Marshal(r)
if err != nil {
return err
}
_, err = js.Publish(ctx, historySubject(r.Module), data, jetstream.WithMsgID(fmt.Sprintf("%s.%d", r.Module, r.Version)))
return err
},
HistoryOf: func(ctx context.Context, module string) ([]meshcatalog.Registration, error) {
cons, err := js.OrderedConsumer(ctx, catalogue.StreamBusName(index.RegistrationHistoryStream), jetstream.OrderedConsumerConfig{FilterSubjects: []string{historySubject(module)}})
if err != nil {
return nil, err
}
var out []meshcatalog.Registration
for {
batch, err := cons.FetchNoWait(64)
if err != nil {
return out, err
}
n := 0
for m := range batch.Messages() {
var r meshcatalog.Registration
if err := json.Unmarshal(m.Data(), &r); err == nil {
out = append(out, r)
}
n++
}
if batch.Error() != nil {
return out, batch.Error()
}
if n == 0 {
return out, nil
}
}
},
Say: func(_ context.Context, e meshcatalog.RegistrationChangedEvent) error {
return publishAs(conn, cred, seatEventSubject(meshcatalog.EventRegistrationChanged), fmt.Sprintf("%s-%s-%d", meshcatalog.EventRegistrationChanged, e.Module, e.Version), e)
},
Branch: func(repository, branch string) (gitseat.BranchAnswer, error) {
var a gitseat.BranchAnswer
raw, err := call("seat:"+gitseat.Seat+"."+gitseat.VerbBranch, gitseat.BranchInput{Repository: repository, Branch: branch})
if err != nil {
return a, err
}
return a, json.Unmarshal(raw, &a)
},
Ask: func(a oc.Ask) error {
return publishAs(conn, cred, oc.AcceptSubjectAsk(cred.Module), a.ID, a)
},
Cancel: func(id string) error {
return publishAs(conn, cred, oc.AcceptSubjectCancel(cred.Module), "cancel."+id, map[string]string{"id": id})
},
Where: where,
SeedWaits: func(ask string, waiting bool) {
if waiting {
logf("seed-waits: the first registrations wait on the operator (ask %s); indexer_seed_again asks them again", ask)
} else {
logf("seed-waits is cleared: the first registrations are asked or written (ask %s)", ask)
}
},
Logf: logf}
}
// errRouterRecord is what the read of the router's record answers until the design says how this program reads it (D-37): a
// warrant is then left unapplied and said, never applied on the decided event's body alone (design 47 G §667, Heard).
var errRouterRecord = errors.New("the router's own record of an ask is not readable from the indexer's program yet (design question D-37)")
func routerRecordUnread(string) (string, *oc.Warrant, error) { return "", nil, errRouterRecord }
// heardDecided is the handler of the decided events for this module: a warrant is applied by the registry; one it could not
// apply is left unacknowledged, so the bus gives it again, and said.
func heardDecided(ctx context.Context, reg *index.Registry, m jetstream.Msg, logf func(string, ...any)) {
var w oc.Warrant
if err := json.Unmarshal(m.Data(), &w); err != nil {
logf("%s holds what is no warrant, dropped: %v", m.Subject(), err)
_ = m.Term()
return
}
if err := reg.Heard(ctx, w); err != nil {
logf("the warrant of the ask %s is left unapplied and given again: %s", w.Ask, strings.TrimSpace(err.Error()))
return
}
_ = m.Ack()
}
@@ -8,34 +8,19 @@ import (
"git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/npmpackageregistry"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"mesh-delivery-indexer/internal/index"
)
// The versions the packages state keeps per series (index.HearVersion, DLV-I5), heard on the indexer's durable consumer of
// the shared events stream: the package registry seats' published (their contracts in mesh-contracts) and the builder's
// numbered (its own event, version.numbered).
// the shared events stream: the package registry seats' published (their contracts in mesh-contracts).
var (
publishedSeats = map[string]string{
gopackageregistry.SubjectPublished: gopackageregistry.Seat,
npmpackageregistry.SubjectPublished: npmpackageregistry.Seat,
}
numberedSubject = "mesh.mod." + builds.BuilderModule + ".event.version.numbered"
)
// numbered is the builder's version.numbered body as the indexer reads it.
type numbered struct {
Name string `json:"name"`
Kind string `json:"kind"`
Version string `json:"version"`
Tag string `json:"tag,omitempty"`
Fingerprint string `json:"fingerprint"`
Commit string `json:"commit,omitempty"`
Digest string `json:"digest,omitempty"`
Seat string `json:"seat,omitempty"`
}
// heardVersion reads one message of the consumer as a version, false for any other subject. A package's is kept in its
// series; one that does not decode is said and acknowledged; a write that fails is left unacknowledged, so the bus gives
// it again.
@@ -55,19 +40,6 @@ func heardVersion(ctx context.Context, ix *index.Indexer, m jetstream.Msg, logf
}
h = index.HeardVersion{Package: p.Name, Seat: seat, Version: p.Version, Tag: p.Tag, Commit: p.Commit, Digest: p.Digest,
Fingerprint: p.Fingerprint, GoModule: p.Module, Published: true}
case m.Subject() == numberedSubject:
var n numbered
if err := json.Unmarshal(m.Data(), &n); err != nil {
logf("%s holds what is no numbered, left: %v", m.Subject(), err)
_ = m.Ack()
return true
}
if n.Kind != "package" {
_ = m.Ack()
return true
}
h = index.HeardVersion{Package: n.Name, Seat: n.Seat, Version: n.Version, Tag: n.Tag, Commit: n.Commit, Digest: n.Digest,
Fingerprint: n.Fingerprint}
default:
return false
}
@@ -27,6 +27,7 @@ type takerIn interface {
FingerprintsAgain(ctx context.Context, module string) index.TakenIn
RecomputeByHand(ctx context.Context, by, why string) index.TakenIn
DropCandidateGraph(ctx context.Context, group, heads, by, why string) index.TakenIn
RegistryAct(ctx context.Context, a index.TakeInAsk) index.TakenIn
}
// takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the
@@ -45,7 +46,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
return
}
asked, r := "take-in", index.TakenIn{}
if a.Recompute {
if a.Registry != nil {
asked, r = "registrations' act "+a.Registry.Verb, ix.RegistryAct(ctx, a)
} else if a.Recompute {
asked, r = "recompute of the catalog graph", ix.RecomputeByHand(ctx, a.By, a.Why)
a.Module = r.Module
} else if a.CandidateGroup != "" && a.Drop {
+3 -1
View File
@@ -9,8 +9,8 @@ require (
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/git v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker v0.0.0-00010101000000-000000000000
@@ -70,6 +70,8 @@ require (
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry v0.0.0-00010101000000-000000000000 // indirect
@@ -0,0 +1,95 @@
package index
import (
"context"
"maps"
"slices"
"strings"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/sdk/packagename"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
// Each entry carries its artifacts' package names (design 47 G §676, §491, §478, §504, §412): at take-in the indexer reads
// each artifact's package name from the revision's own files at the artifact's path, the module of its go.mod for a Go
// package, the name of its package.json for an npm package, the manifest's name of an image artifact, and writes it into
// the entry's artifact as package, {ecosystem, name}. The manifest validator judges §491 from these fields.
// Package ecosystems of an artifact's package (§676).
const (
EcosystemGo = "go"
EcosystemNpm = "npm"
)
// ArtifactRefOf is the artifact reference of a module's artifact, <module>:<artifact>.
func ArtifactRefOf(module, artifact string) string { return module + ":" + artifact }
// withArtifacts gives e its manifest's artifacts, by reference, with the package name of each package artifact (an image's name is an open design question: none is written). A
// name the forge does not answer is said and left out of the artifact, for the next entry to read again (the cache keeps only
// what was read); a manifest that does not decode leaves the entry as it is.
func (ix *Indexer) withArtifacts(ctx context.Context, e meshcatalog.CatalogEntry, m catalogue.Manifest) meshcatalog.CatalogEntry {
if m.Build == nil {
return e
}
for _, a := range m.Build.Artifacts {
art := meshcatalog.Artifact{Ref: ArtifactRefOf(e.Module, a.Name), Language: a.Language, Kind: a.Kind}
if pkg := ix.packageOf(ctx, e, a); pkg != nil {
art.Package = pkg
}
e.Artifacts = append(e.Artifacts, art)
}
slices.SortFunc(e.Artifacts, func(a, b meshcatalog.Artifact) int { return strings.Compare(a.Ref, b.Ref) })
return e
}
// packageOf is the package name of artifact a of the entry e: read through the forge at the entry's commit, kept per
// repository, commit, path and artifact so a later entry of the same commit reads nothing; nil for an artifact that names
// none (a program, an image, an archive) or whose name cannot be read now.
func (ix *Indexer) packageOf(ctx context.Context, e meshcatalog.CatalogEntry, a catalogue.Artifact) *meshcatalog.ArtifactPackage {
switch a.Kind {
case catalogue.ArtifactPackage:
default:
return nil
}
if ix.ReadFile == nil || e.Commit == "" {
return nil
}
ecosystem := EcosystemNpm
if a.Language == "go" {
ecosystem = EcosystemGo
}
key := e.Repository + "\x00" + e.Commit + "\x00" + e.Path + "\x00" + a.Name
ix.namesMu.Lock()
name, ok := ix.packageNames[key]
ix.namesMu.Unlock()
if !ok {
var err error
if name, err = packagename.Of(ctx, packagename.ReadFile(ix.ReadFile), e.Repository, e.Commit, e.Path, a); err != nil {
ix.logf("catalog graph: the package name of %s at %.12s is not read, so its entry carries none until the next entry: %v", ArtifactRefOf(e.Module, a.Name), e.Commit, err)
return nil
}
ix.namesMu.Lock()
if ix.packageNames == nil {
ix.packageNames = map[string]string{}
}
ix.packageNames[key] = name
ix.namesMu.Unlock()
}
return &meshcatalog.ArtifactPackage{Ecosystem: ecosystem, Name: name}
}
// packageEntries is the entry of each module whose manifest decodes, with its artifacts' package names as the take-in reads
// them, for the judge of §491 to read the names from.
func (ix *Indexer) packageEntries(ctx context.Context, held map[string]meshcatalog.ModulesByModuleValue, raws map[string][]byte) []meshcatalog.CatalogEntry {
var out []meshcatalog.CatalogEntry
for _, name := range slices.Sorted(maps.Keys(raws)) {
m, err := catalogue.Decode(raws[name])
if err != nil {
continue
}
cur := held[name]
out = append(out, ix.withArtifacts(ctx, meshcatalog.CatalogEntry{Module: name, Repository: cur.Repository, Path: cur.Path, Commit: cur.Commit}, m))
}
return out
}
@@ -100,11 +100,17 @@ type Indexer struct {
// Revisions is the object bucket catalog-revisions (catalog_revisions.go; design 47, Two graphs); nil keeps no
// catalog graph revision.
Revisions Objects
// EntryIndex is the state entry-index (entry_verb.go, design 47 G §675); nil keeps none.
EntryIndex Store
// SayCatalogRevision says the seat mesh-catalog's event catalog-revision for a revision kept; nil says nothing.
SayCatalogRevision func(ctx context.Context, e meshcatalog.CatalogRevisionEvent) error
// SayFill says the seat mesh-catalog's event filling, full or room of the catalog graph revisions, under the duplicate
// id given (W4.6).
SayFill func(ctx context.Context, event, id string, f meshcatalog.Fill) error
// Registry is the registrations and their trusted changes (registrations.go, design 47 G §2 to §4, §667); SeedModules reads
// what the first registrations are composed from (§8). Nil: no registration act is answered.
Registry *Registry
SeedModules func(ctx context.Context) ([]SeedModule, error)
// Graph is the module graph's state (GraphBucket, graph.go; novox/hq ADR 0603 §2); nil writes no graph.
Graph Store
// Files asks the forge for every blob under each directory at a commit, by directory then path relative to it, each
@@ -159,17 +165,19 @@ type Indexer struct {
saidAt map[string]string // the head of a repository whose failed read was said, by repository: said once per head
failedAt map[string]string // the head of a repository whose tree lacks a directory asked, by repository: not asked again until the head changes
// OwnBuild answers the fingerprint of the build this program runs, from the node-engine's seat; nil knows none.
OwnBuild func(ctx context.Context) (string, error)
underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go)
judgeMu sync.Mutex // one judge pass at a time (verdicts.go)
ownMu sync.Mutex // guards ownFP
keepMu sync.Mutex // one catalog graph revision kept at a time (keepEntries)
catchingUp atomic.Bool // the first start's catch-up runs: its merges keep no revision until it ends (CatchUp)
catchUpWG sync.WaitGroup // the first start's catch-up in the background (kickCatchUp), for tests to wait on
ownFP string // the fingerprint of the build this program runs, fixed once at the start (FixOwnBuild)
ownFixed bool // FixOwnBuild asked already: it asks once
ownSince time.Time // when the build was asked and not known, with ownWhy
ownWhy string
OwnBuild func(ctx context.Context) (string, error)
underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go)
judgeMu sync.Mutex // one judge pass at a time (verdicts.go)
ownMu sync.Mutex // guards ownFP
keepMu sync.Mutex // one catalog graph revision kept at a time (keepEntries)
namesMu sync.Mutex // packageNames
packageNames map[string]string // the package names read through the forge, by repository, commit, path and artifact (artifacts.go)
catchingUp atomic.Bool // the first start's catch-up runs: its merges keep no revision until it ends (CatchUp)
catchUpWG sync.WaitGroup // the first start's catch-up in the background (kickCatchUp), for tests to wait on
ownFP string // the fingerprint of the build this program runs, fixed once at the start (FixOwnBuild)
ownFixed bool // FixOwnBuild asked already: it asks once
ownSince time.Time // when the build was asked and not known, with ownWhy
ownWhy string
// OwnBuildState keeps which build this process knows it runs, or since when and why not (FixOwnBuild); nil keeps none.
OwnBuildState Store
judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go)
@@ -8,6 +8,7 @@ import (
"maps"
"reflect"
"slices"
"strings"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/sdk/graph"
@@ -23,6 +24,24 @@ import (
// CatalogEntryLayout is the layout version an entry and a revision are written in.
const CatalogEntryLayout = 1
// manifestDigestPrefix is how an entry's field manifest-digest begins, as every digest a record names (design 47 G §578).
const manifestDigestPrefix = "sha256:"
// EntryManifestDigest is the entry's field manifest-digest for the module's digest bare, the 64 hex digits the module's
// value and the key of `manifests` hold: `sha256:` and the digits (design 47 G §578). A digest already prefixed stays, and
// none stays none.
func EntryManifestDigest(bare string) string {
if bare == "" || strings.HasPrefix(bare, manifestDigestPrefix) {
return bare
}
return manifestDigestPrefix + bare
}
// BareManifestDigest is the 64 hex digits of an entry's manifest-digest: the module's value and the key of `manifests`
// (design 47 G §578). An entry taken in before G carries the digits alone, which is what its module value held (§609), and
// stays as it is.
func BareManifestDigest(field string) string { return strings.TrimPrefix(field, manifestDigestPrefix) }
// CatalogEntryOf is the catalog graph entry of one module: the module graph's node of it as a recompute computed it
// (its build, provision and verb edges, the verb edges left out on a cycle, ADR 0614), with where the trunk holds it,
// its manifest's digest and verdict, the main package its process runs (MainOf), the package versions it builds against
@@ -30,7 +49,7 @@ const CatalogEntryLayout = 1
// ModuleValueOf gives the module's value back from the entry alone. The acting rules are the revision's to add.
func CatalogEntryOf(module string, cur meshcatalog.ModulesByModuleValue, m catalogue.Manifest, raw string, n graph.Node) meshcatalog.CatalogEntry {
e := meshcatalog.CatalogEntry{V: CatalogEntryLayout, Module: module, Repository: cur.Repository, Path: cur.Path,
Commit: cur.Commit, Manifest: manifestWhole(raw), ManifestDigest: cur.Manifest, Verdict: cur.Verdict, Fp: n.FP, None: n.None, Bases: n.Bases,
Commit: cur.Commit, Manifest: manifestWhole(raw), ManifestDigest: EntryManifestDigest(cur.Manifest), Verdict: cur.Verdict, Fp: n.FP, None: n.None, Bases: n.Bases,
Packages: maps.Clone(cur.Packages), Provides: n.Provides, Requires: n.Requires, Awaits: n.Awaits, Invokes: n.Invokes,
Serves: n.Serves, LeftOut: n.LeftOut, Main: MainOf(m)}
if n.FP == "" && n.None == "" {
@@ -60,7 +79,7 @@ func CatalogEntryOf(module string, cur meshcatalog.ModulesByModuleValue, m catal
// ruling on milestone 47/G finding 1; ADR 0656 §6: one form of the fact). The old line's fields (the old-form
// fingerprint, the reader needs) are not kept.
func ModuleValueOf(e meshcatalog.CatalogEntry) meshcatalog.ModulesByModuleValue {
v := meshcatalog.ModulesByModuleValue{V: keys.Version, Repository: e.Repository, Path: e.Path, Commit: e.Commit, Manifest: e.ManifestDigest,
v := meshcatalog.ModulesByModuleValue{V: keys.Version, Repository: e.Repository, Path: e.Path, Commit: e.Commit, Manifest: BareManifestDigest(e.ManifestDigest),
Verdict: e.Verdict, Fp2: e.Fp, None2: e.None, Packages: maps.Clone(e.Packages)}
if in := e.Inputs; in != nil {
v.Toolchains, v.Paths, v.Problems, v.Warnings, v.Unknown = maps.Clone(in.Toolchains), maps.Clone(in.Paths), in.Problems, in.Warnings, in.Unknown
@@ -1,6 +1,8 @@
package index
import (
"context"
"reflect"
"testing"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
@@ -16,10 +18,53 @@ func TestAnEntryCarriesItsManifestWholeAndTheDigestOfItsBytes(t *testing.T) {
digest := meshcatalogclient.ManifestDigest([]byte(raw))
cur := meshcatalog.ModulesByModuleValue{Repository: "novox/mesh-catalog", Path: "modules/avahi", Commit: "c", Manifest: digest}
e := CatalogEntryOf("avahi", cur, catalogue.Manifest{}, raw, graph.Node{})
if e.ManifestDigest != digest || e.Manifest["module"] != "avahi" {
t.Fatalf("entry digest %q manifest %v, want the digest %s of the bytes and the manifest whole", e.ManifestDigest, e.Manifest, digest)
if e.ManifestDigest != "sha256:"+digest || e.Manifest["module"] != "avahi" {
t.Fatalf("entry digest %q manifest %v, want sha256: and the digits %s of the bytes (§578) and the manifest whole", e.ManifestDigest, e.Manifest, digest)
}
if back := ModuleValueOf(e); back.Manifest != digest {
t.Errorf("the module's value read back holds %q, want the entry's digest %s", back.Manifest, digest)
t.Errorf("the module's value read back holds %q, want the bare digits %s (§578)", back.Manifest, digest)
}
// §609: an entry taken in before G carries the digits its module value held; reading it back gives the same digits.
before := e
before.ManifestDigest = digest
if back := ModuleValueOf(before); back.Manifest != digest {
t.Errorf("an entry of before G read back holds %q, want %s (§609)", back.Manifest, digest)
}
}
// Design 47 G §676: at take-in the entry's artifacts carry the package name read from the revision's own files at the
// artifact's path: the module of a go.mod, the name of a package.json.
func TestAnEntryCarriesItsArtifactsPackageNames(t *testing.T) {
ix := &Indexer{ReadFile: func(_ context.Context, repository, sha, path string) ([]byte, error) {
files := map[string]string{
"mods/m/go/go.mod": "module git.example/m/go\n\ngo 1.26\n",
"mods/m/js/package.json": `{"name":"@example/m"}`,
}
if repository != "r" || sha != "c1" {
t.Errorf("read %s %s %s, want the entry's repository and commit", repository, sha, path)
}
if f, ok := files[path]; ok {
return []byte(f), nil
}
return nil, nil
}}
m := catalogue.Manifest{Module: "m", Build: &catalogue.Build{Artifacts: []catalogue.Artifact{
{Name: "gopkg", Kind: catalogue.ArtifactPackage, Language: "go", Paths: []string{"go"}},
{Name: "jspkg", Kind: catalogue.ArtifactPackage, Language: "typescript", Paths: []string{"js"}},
{Name: "img", Kind: catalogue.ArtifactImage},
{Name: "prog", Kind: "bundle", Language: "go"},
}}}
e := ix.withArtifacts(context.Background(), meshcatalog.CatalogEntry{Module: "m", Repository: "r", Path: "mods/m", Commit: "c1"}, m)
got := map[string]string{}
for _, a := range e.Artifacts {
name := "none"
if a.Package != nil {
name = a.Package.Ecosystem + " " + a.Package.Name
}
got[a.Ref] = name
}
want := map[string]string{"m:gopkg": "go git.example/m/go", "m:jspkg": "npm @example/m", "m:img": "none", "m:prog": "none"}
if !reflect.DeepEqual(got, want) {
t.Errorf("artifacts' packages %v, want %v (§676)", got, want)
}
}
@@ -189,11 +189,11 @@ func (ix *Indexer) keepTrunkRevision(ctx context.Context, edges map[string]graph
// Every module the view holds is an entry, one the recompute computed no edges for (a manifest it cannot read) too.
entries := make(map[string]meshcatalog.CatalogEntry, len(held))
for name, cur := range held {
entries[name] = CatalogEntryOf(name, cur, edges[name].manifest, ix.manifestBytes(ctx, cur.Manifest), whole[name])
entries[name] = ix.withArtifacts(ctx, CatalogEntryOf(name, cur, edges[name].manifest, ix.manifestBytes(ctx, cur.Manifest), whole[name]), edges[name].manifest)
}
for name, e := range edges {
if _, ok := held[name]; !ok {
entries[name] = CatalogEntryOf(name, e.cur, e.manifest, ix.manifestBytes(ctx, e.cur.Manifest), whole[name])
entries[name] = ix.withArtifacts(ctx, CatalogEntryOf(name, e.cur, e.manifest, ix.manifestBytes(ctx, e.cur.Manifest), whole[name]), e.manifest)
}
}
return ix.keepEntries(ctx, entries)
@@ -235,7 +235,7 @@ func (ix *Indexer) KeepView(ctx context.Context) error {
m, raw = read, bytes
}
}
entries[name] = CatalogEntryOf(name, cur, m, raw, nodes[name])
entries[name] = ix.withArtifacts(ctx, CatalogEntryOf(name, cur, m, raw, nodes[name]), m)
}
return ix.keepEntries(ctx, entries)
}
@@ -278,6 +278,10 @@ func (ix *Indexer) keepEntries(ctx context.Context, entries map[string]meshcatal
ix.sayFill(ctx, meshcatalog.EventFull, fmt.Sprintf("%d", src), meshcatalog.Fill{Bytes: used, Cap: capacity, Refused: refused})
return fmt.Errorf("revision %d of source revision %d: %w", rev.Revision, src, ErrObjectsFull)
}
if err := ix.indexEntries(ctx, rev, entries); err != nil {
ix.logf("catalog graph: revision %d of source revision %d is not kept: %v", rev.Revision, src, err)
return err
}
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
ix.logf("catalog graph: revision %d of source revision %d is not kept: %v", rev.Revision, src, err)
return err
@@ -0,0 +1,143 @@
package index
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/process"
)
// The seat mesh-catalog's verb entry finds a catalog graph entry by its build-input fingerprint (design 47 G §675, §443):
// the indexer keeps, in its old state section entry-index, one key per fingerprint holding the digest of its entry's
// object, written when it stores the entry; entry reads the key and then the object. The state is capped at 8 MiB.
const (
// EntryIndexState is the old state section of the entries' digests by fingerprint.
EntryIndexState = "entry-index"
// EntryIndexBucket is the bucket of the state.
EntryIndexBucket = Module + "_" + EntryIndexState
// EntryIndexTool is the tool that reads a fingerprint's key and its object's digest (class read).
EntryIndexTool = "indexer_entries"
)
// EntryIndexValue is one key of entry-index: the digest of the entry object of the fingerprint the key names.
type EntryIndexValue struct {
V int `json:"v"`
Digest string `json:"digest"`
}
// EntryIndexKey is the key of a fingerprint in entry-index, the fingerprint escaped as a key's name is (§73).
func EntryIndexKey(fingerprint string) string { return keys.EscapeName(fingerprint) }
// indexEntries writes the entry-index key of each entry of rev that has a fingerprint, the digest of its object; a key that
// already holds that digest is not written again. A fingerprint two entries carry holds the later module's, which is the
// newest entry written.
func (ix *Indexer) indexEntries(ctx context.Context, rev meshcatalog.CatalogRevision, entries map[string]meshcatalog.CatalogEntry) error {
if ix.EntryIndex == nil {
return nil
}
var errs []error
for _, ref := range rev.Entries {
fp := entries[ref.Module].Fp
if fp == "" {
continue
}
data, err := json.Marshal(EntryIndexValue{V: keys.Version, Digest: ref.Digest})
if err != nil {
return err
}
err = write(ctx, ix.EntryIndex, EntryIndexKey(fp), func([]byte, bool) ([]byte, bool, error) { return data, false, nil })
if errors.Is(err, process.ErrNotYet) {
ix.logf("catalog graph: the state %s is not on the bus yet, so the entries of this revision are not found by their fingerprint", EntryIndexState)
return nil
}
if err != nil {
errs = append(errs, fmt.Errorf("entry-index of %s: %w", ref.Module, err))
}
}
return errors.Join(errs...)
}
// Entry answers the verb entry: the entry whose key entry-index holds for the fingerprint, read from its object.
func Entry(ctx context.Context, o Objects, index Store, in meshcatalog.EntryInput) (meshcatalog.EntryAnswer, error) {
var out meshcatalog.EntryAnswer
if in.Fingerprint == "" {
return out, errors.New("name the entry's fingerprint (fingerprint)")
}
value, _, found, err := index.Get(ctx, EntryIndexKey(in.Fingerprint))
if err != nil {
return out, fmt.Errorf("the state %s cannot be read: %w", EntryIndexState, err)
}
var v EntryIndexValue
if !found || json.Unmarshal(value, &v) != nil || v.Digest == "" {
return out, fmt.Errorf("no entry of fingerprint %s is kept", in.Fingerprint)
}
data, err := o.Get(ctx, EntryObject(v.Digest))
if errors.Is(err, ErrNoObject) {
return out, fmt.Errorf("the entry object %s of fingerprint %s is not kept", v.Digest, in.Fingerprint)
}
if err != nil {
return out, fmt.Errorf("the entry object %s: %w", v.Digest, err)
}
if err := json.Unmarshal(data, &out.Entry); err != nil {
return out, fmt.Errorf("the entry object %s cannot be read: %w", v.Digest, err)
}
return out, nil
}
// EntryTool is the verb entry as the tool mesh-catalog.entry of the indexer's own runtime, as RevisionTool is revision.
func EntryTool(o Objects, index Store) stdio.Tool {
return stdio.Tool{Name: meshcatalog.Seat + "." + meshcatalog.VerbEntry, Run: func(args map[string]any) (any, error) {
var in meshcatalog.EntryInput
if raw, err := json.Marshal(args); err == nil {
_ = json.Unmarshal(raw, &in)
}
return Entry(context.Background(), o, index, in)
}}
}
// EntriesAnswer is what indexer_entries answers: for a fingerprint its key and its object's digest, and whether the object
// is kept; with none given, every fingerprint the state holds.
type EntriesAnswer struct {
Fingerprint string `json:"fingerprint,omitempty"`
Digest string `json:"digest,omitempty"`
Kept bool `json:"kept,omitempty"`
Keys []string `json:"keys,omitempty"`
}
// EntriesTool is indexer_entries (class read): one fingerprint's key and the digest of its object, or every key of
// entry-index with none given.
func EntriesTool(o Objects, index Store) stdio.Tool {
return stdio.Tool{Name: EntryIndexTool, Description: "Read the indexer's entry-index: the digest of the catalog graph entry of a fingerprint and whether its object is kept; with no fingerprint, every key it holds.",
Input: map[string]any{"type": "object", "properties": map[string]any{"fingerprint": map[string]any{"type": "string", "description": "the entry's fingerprint; none lists every key"}}},
Run: func(args map[string]any) (any, error) {
ctx := context.Background()
fp, _ := args["fingerprint"].(string)
if fp == "" {
all, err := index.Keys(ctx)
if err != nil {
return nil, err
}
slices.Sort(all)
return EntriesAnswer{Keys: all}, nil
}
out := EntriesAnswer{Fingerprint: fp}
value, _, found, err := index.Get(ctx, EntryIndexKey(fp))
if err != nil {
return nil, err
}
var v EntryIndexValue
if found && json.Unmarshal(value, &v) == nil {
out.Digest = v.Digest
if _, err := o.Get(ctx, EntryObject(v.Digest)); err == nil {
out.Kept = true
}
}
return out, nil
}}
}
@@ -0,0 +1,58 @@
package index
import (
"context"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// Design 47 G §675: the entry kept for a build-input fingerprint is found by it, through the key entry-index holds for the
// digest of its object; a fingerprint no entry carries is answered with an error naming it.
func TestEntryFindsAnEntryByItsFingerprint(t *testing.T) {
ctx := context.Background()
fp := "src2:" + strings.Repeat("b", 64)
ix := &Indexer{Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
entries := map[string]meshcatalog.CatalogEntry{
"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c", Fp: fp, Manifest: map[string]any{}},
"none": {Module: "none", Repository: "novox/r", Commit: "c", Manifest: map[string]any{}},
}
rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: 1, Kind: "trunk"}, entries)
if err != nil {
t.Fatal(err)
}
if err := ix.indexEntries(ctx, rev, entries); err != nil {
t.Fatal(err)
}
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
t.Fatal(err)
}
if got, err := Entry(ctx, ix.Revisions, ix.EntryIndex, meshcatalog.EntryInput{Fingerprint: fp}); err != nil || got.Entry.Module != "avahi" {
t.Fatalf("entry of %s: %+v %v, want avahi's", fp, got, err)
}
if keys, _ := ix.EntryIndex.Keys(ctx); len(keys) != 1 {
t.Errorf("entry-index holds %v, want the one key of the entry that has a fingerprint", keys)
}
if _, err := Entry(ctx, ix.Revisions, ix.EntryIndex, meshcatalog.EntryInput{Fingerprint: "src2:" + strings.Repeat("c", 64)}); err == nil || !strings.Contains(err.Error(), "no entry of fingerprint") {
t.Errorf("an unknown fingerprint: %v, want an error naming it", err)
}
if _, err := Entry(ctx, ix.Revisions, ix.EntryIndex, meshcatalog.EntryInput{}); err == nil {
t.Errorf("no fingerprint was answered")
}
}
// Design 47 G §675: the revision the indexer keeps writes entry-index for every entry that has a fingerprint, before it keeps
// the revision, so entry finds the entry as soon as the revision is kept.
func TestAKeptRevisionIndexesItsEntries(t *testing.T) {
ctx := context.Background()
fp := "src2:" + strings.Repeat("d", 64)
ix := &Indexer{Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
if err := ix.keepEntries(ctx, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c", Fp: fp, Manifest: map[string]any{}}}); err != nil {
t.Fatal(err)
}
if got, err := Entry(ctx, ix.Revisions, ix.EntryIndex, meshcatalog.EntryInput{Fingerprint: fp}); err != nil || got.Entry.Module != "avahi" {
t.Fatalf("entry of %s after the revision was kept: %+v %v", fp, got, err)
}
}
@@ -0,0 +1,133 @@
package index
import (
"bytes"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"sort"
"strings"
"unicode/utf16"
)
// Canonical is a JSON value in RFC 8785's canonical form: keys sorted by their UTF-16 code units, no white space, strings
// escaped as the RFC says, integers as digits. A number with a fraction or an exponent is refused: no change the indexer
// digests holds one.
func Canonical(v any) ([]byte, error) {
raw, err := json.Marshal(v)
if err != nil {
return nil, err
}
dec := json.NewDecoder(bytes.NewReader(raw))
dec.UseNumber()
var tree any
if err := dec.Decode(&tree); err != nil {
return nil, err
}
var b bytes.Buffer
if err := canonical(&b, tree); err != nil {
return nil, err
}
return b.Bytes(), nil
}
// DigestOf is "sha256:" and the digits of the canonical form of v.
func DigestOf(v any) (string, error) {
c, err := Canonical(v)
if err != nil {
return "", err
}
sum := sha256.Sum256(c)
return "sha256:" + hex.EncodeToString(sum[:]), nil
}
func canonical(b *bytes.Buffer, v any) error {
switch t := v.(type) {
case nil:
b.WriteString("null")
case bool:
if t {
b.WriteString("true")
} else {
b.WriteString("false")
}
case json.Number:
if strings.ContainsAny(string(t), ".eE") {
return fmt.Errorf("the number %s has a fraction or an exponent, which the canonical form of a change does not hold", t)
}
b.WriteString(string(t))
case string:
canonicalString(b, t)
case []any:
b.WriteByte('[')
for i, e := range t {
if i > 0 {
b.WriteByte(',')
}
if err := canonical(b, e); err != nil {
return err
}
}
b.WriteByte(']')
case map[string]any:
keys := make([]string, 0, len(t))
for k := range t {
keys = append(keys, k)
}
sort.Slice(keys, func(i, j int) bool { return slices16(keys[i], keys[j]) })
b.WriteByte('{')
for i, k := range keys {
if i > 0 {
b.WriteByte(',')
}
canonicalString(b, k)
b.WriteByte(':')
if err := canonical(b, t[k]); err != nil {
return err
}
}
b.WriteByte('}')
default:
return fmt.Errorf("a value of type %T has no canonical form", v)
}
return nil
}
// slices16 orders two strings by their UTF-16 code units, as RFC 8785 sorts keys.
func slices16(a, b string) bool {
x, y := utf16.Encode([]rune(a)), utf16.Encode([]rune(b))
for i := 0; i < len(x) && i < len(y); i++ {
if x[i] != y[i] {
return x[i] < y[i]
}
}
return len(x) < len(y)
}
func canonicalString(b *bytes.Buffer, s string) {
b.WriteByte('"')
for _, r := range s {
switch {
case r == '"':
b.WriteString(`\"`)
case r == '\\':
b.WriteString(`\\`)
case r == '\b':
b.WriteString(`\b`)
case r == '\f':
b.WriteString(`\f`)
case r == '\n':
b.WriteString(`\n`)
case r == '\r':
b.WriteString(`\r`)
case r == '\t':
b.WriteString(`\t`)
case r < 0x20:
fmt.Fprintf(b, `\u%04x`, r)
default:
b.WriteRune(r)
}
}
b.WriteByte('"')
}
@@ -910,7 +910,9 @@ func (ix *Indexer) dry() *dryRun {
Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil,
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}
Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil,
// A dry run writes no registration and asks no one.
Registry: nil, SeedModules: nil}
x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) }
x.Store, d.candidates = over(ix.Store)
x.Modules, d.modules = over(ix.Modules)
@@ -1133,7 +1133,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", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill"} {
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules"} {
named[f] = true
}
ty := reflect.TypeOf(Indexer{})
@@ -2,6 +2,7 @@ package index
import (
"context"
"encoding/json"
"fmt"
"maps"
"slices"
@@ -224,6 +225,8 @@ type TakenIn struct {
Unread []string `json:"unread,omitempty"`
// Fingerprints are the modules whose fingerprint a recompute by hand moved (FingerprintsAgain), each old and new.
Fingerprints []FingerprintMove `json:"fingerprints,omitempty"`
// Registry is the answer of a registrations' act (RegistryAct): the verb's own answer, as JSON.
Registry json.RawMessage `json:"registry,omitempty"`
}
// TakeIn reads a registered module's manifest now, whatever was tried before, and writes it as a merge would (the tool
@@ -0,0 +1,921 @@
package index
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"sort"
"strings"
"sync"
"time"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
gitseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/git"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// A module's registration, and the trusted change of one (design 47 G §2 to §4, §10, §342, §343, §667): the indexer keeps each
// module's newest registration in its old state section registrations, one key per module, every version on its old
// streams section's stream registration-history, and writes a version only on the operator's warrant. A change waits for
// its warrant in the old state section proposals.
const (
// RegistrationsState is the old state section of the registrations, one key per module (§2).
RegistrationsState = "registrations"
// RegistrationsBucket is the bucket of the state.
RegistrationsBucket = Module + "_" + RegistrationsState
// ProposalsState is the old state section of the changes that wait for their warrant (§667), capped at 4 MiB.
ProposalsState = "proposals"
// ProposalsBucket is the bucket of the state.
ProposalsBucket = Module + "_" + ProposalsState
// RegistrationHistoryStream is the old streams section that keeps every version of every registration (§2, §9).
RegistrationHistoryStream = "registration-history"
// RegistrationHistoryEvent is the event of the stream, one subject per module.
RegistrationHistoryEvent = "registration"
// SeedKey is the key of the first registrations' proposal (§8).
SeedKey = "seed"
// ProposalsTool, ProposalDropTool and SeedAgainTool are the tools of the proposals (§667, §8).
ProposalsTool = "indexer_proposals"
ProposalDropTool = "indexer_proposal_drop"
SeedAgainTool = "indexer_seed_again"
// ProposalLasts is how long the operator has to answer: the router's bound for an approval less ten minutes.
ProposalLasts = occ.ApproveLasts - 10*time.Minute
)
// The states of a proposal's key (§667): open while its warrant is awaited, then how it ended.
const (
ProposalOpen = "open"
ProposalApplied = "applied"
ProposalDeclined = "declined"
ProposalExpired = "expired"
ProposalRefused = "refused"
ProposalDropped = "dropped"
ProposalLost = "lost"
// ProposalSeedWaits is the seed's end that keeps the finding seed-waits (§8).
ProposalSeedWaits = "seed-waits"
)
const (
optionApprove = "approve"
optionDecline = "decline"
)
// The verbs of a trusted change.
const (
verbRegister = "register"
verbPolicy = "policy"
verbWatch = "watch"
verbUnwatch = "unwatch"
verbSeed = "seed"
)
// Proposal is one value of the state proposals: a change awaiting its warrant, whole, with its digest, the current ask's id,
// its caller and when it was asked; ended, it keeps its outcome (§667).
type Proposal struct {
V int `json:"v"`
Verb string `json:"verb"`
Module string `json:"module,omitempty"`
Branch string `json:"branch,omitempty"`
Change json.RawMessage `json:"change"`
// Was is the version of the registration the change was composed against (0 for none): a change warranted when the
// registration moved is refused, never written over the newer one.
Was int64 `json:"was"`
Lines []string `json:"lines"`
Digest string `json:"digest"`
Ask string `json:"ask"`
Caller string `json:"caller,omitempty"`
Why string `json:"why,omitempty"`
At time.Time `json:"at"`
// State is open, or how the key ended; Outcome says it in words; Ended is when.
State string `json:"state"`
Outcome string `json:"outcome,omitempty"`
Ended time.Time `json:"ended,omitzero"`
// Seed holds, for the seed's key, the registrations it proposes (§8).
Seed []meshcatalog.Registration `json:"seed,omitempty"`
}
// ProposalKey is the key of a change: <verb>.<module>, and for watch and unwatch <verb>.<module>.<branch>, each token
// escaped as a key's name is (§667).
func ProposalKey(verb, module, branch string) string {
k := verb + "." + keys.EscapeName(module)
if branch != "" {
k += "." + keys.EscapeName(branch)
}
return k
}
// RegistrationKey is a module's key in registrations.
func RegistrationKey(module string) string { return keys.EscapeName(module) }
// Registry is the registrations and the trusted changes of them. Its parts are the indexer's: the two states, the history
// stream, the event, the seat git's branch, the seat operator-channel's ask.
type Registry struct {
Registrations Store
Proposals Store
// History keeps one version on the stream registration-history, once under its key <module>.<version>; HistoryOf reads
// every version of one module, oldest first.
History func(ctx context.Context, r meshcatalog.Registration) error
HistoryOf func(ctx context.Context, module string) ([]meshcatalog.Registration, error)
// Say says the seat mesh-catalog's event registration-changed.
Say func(ctx context.Context, e meshcatalog.RegistrationChangedEvent) error
// Branch reads a branch through the seat git; an empty branch asks the repository's default branch.
Branch func(repository, branch string) (gitseat.BranchAnswer, error)
// Ask publishes an ask under the indexer's own name, Cancel takes it back, Where reads the router's own record of an
// ask, read under the indexer's module's name (state, and the warrant when it holds one; none when it holds no ask).
Ask func(a oc.Ask) error
Cancel func(id string) error
Where func(id string) (state string, w *oc.Warrant, err error)
// SeedWaits says the finding seed-waits naming the ask, and clears it (§8); nil says nothing.
SeedWaits func(ask string, waiting bool)
// Seed composes the first registrations (§8); nil proposes none.
Seed func(ctx context.Context) (SeedProposal, error)
Now func() time.Time
Logf func(string, ...any)
mu sync.Mutex
}
func (r *Registry) now() time.Time {
if r.Now != nil {
return r.Now().UTC()
}
return time.Now().UTC()
}
func (r *Registry) logf(format string, args ...any) {
if r.Logf != nil {
r.Logf(format, args...)
}
}
// Registration is one module's newest registration.
func (r *Registry) Registration(ctx context.Context, module string) (meshcatalog.Registration, bool, error) {
v, _, found, err := r.Registrations.Get(ctx, RegistrationKey(module))
if err != nil || !found {
return meshcatalog.Registration{}, false, err
}
var reg meshcatalog.Registration
if err := json.Unmarshal(v, &reg); err != nil {
return reg, false, fmt.Errorf("the registration of %s cannot be read: %w", module, err)
}
return reg, true, nil
}
// Registrations is the verb registrations (§4): every registration, or one module's, with its history.
func (r *Registry) RegistrationsAnswer(ctx context.Context, in meshcatalog.RegistrationsInput) (meshcatalog.RegistrationsAnswer, error) {
out := meshcatalog.RegistrationsAnswer{Registrations: []meshcatalog.Registration{}}
if in.Module != "" {
reg, found, err := r.Registration(ctx, in.Module)
if err != nil {
return out, err
}
if found {
out.Registrations = append(out.Registrations, reg)
}
if in.History && r.HistoryOf != nil {
if out.History, err = r.HistoryOf(ctx, in.Module); err != nil {
return out, fmt.Errorf("the history of %s cannot be read: %w", in.Module, err)
}
}
return out, nil
}
all, err := r.Registrations.Keys(ctx)
if err != nil {
return out, err
}
slices.Sort(all)
for _, k := range all {
v, _, found, err := r.Registrations.Get(ctx, k)
if err != nil {
return out, err
}
var reg meshcatalog.Registration
if found && json.Unmarshal(v, &reg) == nil {
out.Registrations = append(out.Registrations, reg)
}
}
return out, nil
}
// refused is the answer of a change that raises no ask.
func refused(module, reason, words string) meshcatalog.RegistrationAsked {
return meshcatalog.RegistrationAsked{Module: module, Refused: reason, Words: words}
}
// forbiddenBranch refuses the two names the catalog's keys use (§343).
func forbiddenBranch(b string) string {
if b == "candidate" || b == "entry" {
return fmt.Sprintf("the branch %q cannot be watched: the catalog keeps its revisions under <branch>.<revision> beside the key forms candidate and entry", b)
}
return ""
}
// reviewRequired refuses code review as a required step (§3, §482).
func reviewRequired(s meshcatalog.StepStatuses) bool {
return s.Review == meshcatalog.StepStatusesApprovalRequired
}
// protect refuses a watched branch the forge does not protect (§5), naming it.
func (r *Registry) protect(repository string, branches []string) (string, error) {
for _, b := range branches {
if w := forbiddenBranch(b); w != "" {
return w, nil
}
a, err := r.Branch(repository, b)
if err != nil {
return "", fmt.Errorf("the seat git does not answer for %s %s: %w", repository, b, err)
}
if a.Error != "" {
return "", fmt.Errorf("the seat git cannot read %s %s: %s", repository, b, a.Error)
}
if !a.Protected {
return fmt.Sprintf("the branch %s of %s is not protected: no rule refuses a direct push and admits only merges", b, repository), nil
}
}
return "", nil
}
// stepLines names each step status a policy moves, one line each.
func stepLines(prefix string, old, next meshcatalog.StepStatuses) []string {
var lines []string
for _, s := range []struct {
name string
was, now string
}{{"approval", old.Approval, next.Approval}, {"composition", old.Composition, next.Composition},
{"integration", old.Integration, next.Integration}, {"lint", old.Lint, next.Lint},
{"code review", old.Review, next.Review}, {"unit tests", old.Unit, next.Unit}} {
if s.was != s.now {
lines = append(lines, fmt.Sprintf("%s%s: %s -> %s", prefix, s.name, cmpEmpty(s.was), cmpEmpty(s.now)))
}
}
return lines
}
func cmpEmpty(s string) string {
if s == "" {
return "none"
}
return s
}
// overlay puts the statuses a policy names over the ones the module keeps.
func overlay(base, over meshcatalog.StepStatuses) meshcatalog.StepStatuses {
for _, p := range []struct {
b *string
o string
}{{&base.Approval, over.Approval}, {&base.Composition, over.Composition},
{&base.Integration, over.Integration}, {&base.Lint, over.Lint}, {&base.Review, over.Review}, {&base.Unit, over.Unit}} {
if p.o != "" {
*p.b = p.o
}
}
return base
}
// compose is the registration a change makes of the one kept (none: found false), and the lines that say each value it moves.
func compose(verb string, cur meshcatalog.Registration, found bool, change json.RawMessage) (meshcatalog.Registration, []string, error) {
switch verb {
case verbRegister:
var in meshcatalog.RegisterInput
if err := json.Unmarshal(change, &in); err != nil {
return cur, nil, err
}
next := in.Registration
next.Version, next.Warrant, next.By, next.At = 0, "", "", nil
var lines []string
if !found {
lines = append(lines, fmt.Sprintf("%s: registered from %s at %s, watching %s", next.Module, next.Origin, next.Path, strings.Join(next.Branches, ", ")))
lines = append(lines, stepLines(next.Module+" ", meshcatalog.StepStatuses{}, next.Policy.Steps)...)
return next, lines, nil
}
if cur.Origin != next.Origin {
lines = append(lines, fmt.Sprintf("%s: origin %s -> %s", next.Module, cur.Origin, next.Origin))
}
if cur.Path != next.Path {
lines = append(lines, fmt.Sprintf("%s: manifest path %s -> %s", next.Module, cur.Path, next.Path))
}
for _, b := range next.Branches {
if !slices.Contains(cur.Branches, b) {
lines = append(lines, fmt.Sprintf("%s: watches %s", next.Module, b))
}
}
for _, b := range cur.Branches {
if !slices.Contains(next.Branches, b) {
lines = append(lines, fmt.Sprintf("%s: stops watching %s", next.Module, b))
}
}
lines = append(lines, stepLines(next.Module+" ", cur.Policy.Steps, next.Policy.Steps)...)
lines = append(lines, artifactLines(next.Module, cur.Policy, next.Policy)...)
return next, lines, nil
case verbPolicy:
var in meshcatalog.PolicyInput
if err := json.Unmarshal(change, &in); err != nil {
return cur, nil, err
}
next := cur
var lines []string
if in.Artifact != "" {
next.Policy.Artifacts = map[string]meshcatalog.StepStatuses{}
for k, v := range cur.Policy.Artifacts {
next.Policy.Artifacts[k] = v
}
was := cur.Policy.Artifacts[in.Artifact]
now := overlay(was, in.Steps)
next.Policy.Artifacts[in.Artifact] = now
lines = stepLines(fmt.Sprintf("%s:%s ", in.Module, in.Artifact), was, now)
} else {
next.Policy.Steps = overlay(cur.Policy.Steps, in.Steps)
lines = stepLines(in.Module+" ", cur.Policy.Steps, next.Policy.Steps)
}
return next, lines, nil
case verbWatch, verbUnwatch:
var in meshcatalog.WatchInput
if err := json.Unmarshal(change, &in); err != nil {
return cur, nil, err
}
next := cur
next.Branches = slices.Clone(cur.Branches)
var lines []string
if verb == verbWatch && !slices.Contains(cur.Branches, in.Branch) {
next.Branches = append(next.Branches, in.Branch)
lines = append(lines, fmt.Sprintf("%s: watches %s", in.Module, in.Branch))
}
if verb == verbUnwatch && slices.Contains(cur.Branches, in.Branch) {
next.Branches = slices.DeleteFunc(next.Branches, func(b string) bool { return b == in.Branch })
lines = append(lines, fmt.Sprintf("%s: stops watching %s", in.Module, in.Branch))
}
return next, lines, nil
}
return cur, nil, fmt.Errorf("%q is no change of a registration", verb)
}
func artifactLines(module string, old, next meshcatalog.Policy) []string {
names := map[string]bool{}
for k := range old.Artifacts {
names[k] = true
}
for k := range next.Artifacts {
names[k] = true
}
sorted := make([]string, 0, len(names))
for k := range names {
sorted = append(sorted, k)
}
sort.Strings(sorted)
var lines []string
for _, a := range sorted {
lines = append(lines, stepLines(module+":"+a+" ", old.Artifacts[a], next.Artifacts[a])...)
}
return lines
}
// Propose is the common path of the four verbs (§3): it checks the change, composes the lines it moves, keeps it in proposals
// under its key (a newer change of the same key replaces the open ask, ADR 0666 §12) and asks the operator, naming every value
// it moves and carrying the change's digest. It writes no registration.
func (r *Registry) Propose(ctx context.Context, verb, caller string, change json.RawMessage) (meshcatalog.RegistrationAsked, error) {
var module, branch, why string
switch verb {
case verbRegister:
var in meshcatalog.RegisterInput
if err := json.Unmarshal(change, &in); err != nil {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("register takes a registration: %w", err)
}
module, why = in.Registration.Module, in.Why
case verbPolicy:
var in meshcatalog.PolicyInput
if err := json.Unmarshal(change, &in); err != nil {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("policy takes a module and its steps: %w", err)
}
module, why = in.Module, in.Why
case verbWatch, verbUnwatch:
var in meshcatalog.WatchInput
if err := json.Unmarshal(change, &in); err != nil {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("%s takes a module and a branch: %w", verb, err)
}
module, branch, why = in.Module, in.Branch, in.Why
default:
return meshcatalog.RegistrationAsked{}, fmt.Errorf("%q is no change of a registration", verb)
}
if strings.TrimSpace(module) == "" {
return meshcatalog.RegistrationAsked{}, errors.New("name the module (module)")
}
if (verb == verbWatch || verb == verbUnwatch) && branch == "" {
return meshcatalog.RegistrationAsked{}, errors.New("name the branch (branch)")
}
r.mu.Lock()
defer r.mu.Unlock()
cur, found, err := r.Registration(ctx, module)
if err != nil {
return meshcatalog.RegistrationAsked{}, err
}
if verb != verbRegister && !found {
return refused(module, meshcatalog.RegistrationAskedRefusedUnknownModule, module+" is not registered: register it first"), nil
}
// register with no branch takes the repository's default branch (§342).
if verb == verbRegister {
var in meshcatalog.RegisterInput
_ = json.Unmarshal(change, &in)
if in.Registration.Origin == "" || in.Registration.Path == "" {
return meshcatalog.RegistrationAsked{}, errors.New("name the registration's origin and the path of its manifest")
}
if len(in.Registration.Branches) == 0 {
a, err := r.Branch(in.Registration.Origin, "")
if err != nil {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("the seat git does not answer the default branch of %s: %w", in.Registration.Origin, err)
}
if a.Error != "" || a.Branch == "" {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("the default branch of %s is not read: %s", in.Registration.Origin, a.Error)
}
in.Registration.Branches = []string{a.Branch}
if change, err = json.Marshal(in); err != nil {
return meshcatalog.RegistrationAsked{}, err
}
}
}
steps := []meshcatalog.StepStatuses{}
switch verb {
case verbRegister:
var in meshcatalog.RegisterInput
_ = json.Unmarshal(change, &in)
steps = append(steps, in.Registration.Policy.Steps)
for _, a := range in.Registration.Policy.Artifacts {
steps = append(steps, a)
}
case verbPolicy:
var in meshcatalog.PolicyInput
_ = json.Unmarshal(change, &in)
steps = append(steps, in.Steps)
}
for _, s := range steps {
if reviewRequired(s) {
return refused(module, meshcatalog.RegistrationAskedRefusedReviewRequired,
"code review is never a required step: no record binds a forge account to its bus identity, so no review can gate while its author cannot be told"), nil
}
}
next, lines, err := compose(verb, cur, found, change)
if err != nil {
return meshcatalog.RegistrationAsked{}, err
}
var branches []string
switch verb {
case verbRegister:
branches = next.Branches
case verbWatch:
branches = []string{branch}
}
if len(branches) > 0 {
words, err := r.protect(next.Origin, branches)
if err != nil {
return meshcatalog.RegistrationAsked{}, err
}
if words != "" {
return refused(module, meshcatalog.RegistrationAskedRefusedUnprotectedBranch, words), nil
}
}
if len(lines) == 0 {
return refused(module, meshcatalog.RegistrationAskedRefusedNothingMoves, "the change moves no value of "+module+"'s registration"), nil
}
p := Proposal{V: 1, Verb: verb, Module: module, Branch: branch, Change: change, Lines: lines, Caller: caller, Why: why, At: r.now(), State: ProposalOpen}
if found {
p.Was = cur.Version
}
if p.Digest, err = DigestOf(map[string]any{"verb": verb, "module": module, "change": json.RawMessage(change)}); err != nil {
return meshcatalog.RegistrationAsked{}, err
}
if err := r.keepAndAsk(ctx, ProposalKey(verb, module, branch), &p); err != nil {
return meshcatalog.RegistrationAsked{}, err
}
return meshcatalog.RegistrationAsked{Module: module, Ask: p.Ask, Lines: lines}, nil
}
// askID is the id of the ask of a change: it names the key's digest and time, so a newer change of the key is a new id.
func askID(key, digest string, at time.Time) string {
d := strings.TrimPrefix(digest, "sha256:")
if len(d) > 16 {
d = d[:16]
}
return fmt.Sprintf("reg-%s-%d", d, at.Unix())
}
// composeAsk is the ask of a proposal, composed again from it wherever it is checked.
func (p Proposal) composeAsk(key string) oc.Ask {
headline := "Change the registration of " + p.Module + "?"
if p.Verb == verbSeed {
headline = "Register the first modules?"
}
if r := []rune(headline); len(r) > occ.HeadlineLength {
headline = string(r[:occ.HeadlineLength-1]) + "…"
}
explain := strings.Join(p.Lines, "\n")
if p.Why != "" {
explain += "\nWhy: " + p.Why
}
explain += "\nDigest " + p.Digest
a := oc.Ask{ID: p.Ask, Headline: headline, Explanation: explain, Who: occ.Operator,
Expires: occ.FormatTime(p.At.Add(ProposalLasts)), OnExpiry: "nothing is written; the change waits to be proposed again",
About: "catalog.proposal." + key}
for _, o := range []struct{ id, label, does, level string }{
{optionApprove, "Approve", "the registration is written as these lines say", occ.Approve},
{optionDecline, "Decline", "nothing is written", occ.Acknowledge},
} {
opt := oc.Option{ID: o.id, Label: o.label, Does: o.does, Level: o.level}
if o.level != occ.Acknowledge {
opt.Binds, _ = occ.ActDigest(p.act())
}
a.Options = append(a.Options, opt)
}
return a
}
// act is what the Approve binds: the change's verb, its module and its digest (ADR 0286 §10).
func (p Proposal) act() occ.Act {
return occ.Act{"verb": p.Verb, "module": p.Module, "digest": p.Digest}
}
// keepAndAsk keeps the proposal under its key and publishes its ask; a newer change of the key replaces the open ask.
func (r *Registry) keepAndAsk(ctx context.Context, key string, p *Proposal) error {
p.Ask = askID(key, p.Digest, p.At)
ask := p.composeAsk(key)
if err := occ.Check(ask, p.At); err != nil {
return err
}
data, err := json.Marshal(p)
if err != nil {
return err
}
if err := write(ctx, r.Proposals, key, func(old []byte, found bool) ([]byte, bool, error) {
if found {
var o Proposal
if json.Unmarshal(old, &o) == nil && o.State == ProposalOpen && o.Ask == p.Ask {
return nil, false, fmt.Errorf("the same change is asked already, as %s", o.Ask)
}
}
return data, false, nil
}); err != nil {
return fmt.Errorf("the change could not be kept, so the operator was not asked: %w", err)
}
if err := r.Ask(ask); err != nil {
p.State, p.Outcome, p.Ended = ProposalRefused, "the ask could not be published: "+err.Error(), r.now()
_ = r.keep(ctx, key, p)
return fmt.Errorf("the operator could not be asked: %w", err)
}
return nil
}
func (r *Registry) keep(ctx context.Context, key string, p *Proposal) error {
data, err := json.Marshal(p)
if err != nil {
return err
}
return write(ctx, r.Proposals, key, func([]byte, bool) ([]byte, bool, error) { return data, false, nil })
}
// ProposalRow is one key of proposals as indexer_proposals lists it.
type ProposalRow struct {
Key string `json:"key"`
Verb string `json:"verb"`
Module string `json:"module,omitempty"`
State string `json:"state"`
Ask string `json:"ask"`
Caller string `json:"caller,omitempty"`
At time.Time `json:"at"`
Outcome string `json:"outcome,omitempty"`
Ended time.Time `json:"ended,omitzero"`
Lines []string `json:"lines,omitempty"`
Digest string `json:"digest"`
}
// ListProposals is the open and the ended keys with their asks and outcomes (indexer_proposals, class read).
func ListProposals(ctx context.Context, st Store) ([]ProposalRow, error) {
all, err := st.Keys(ctx)
if err != nil {
return nil, err
}
slices.Sort(all)
out := []ProposalRow{}
for _, k := range all {
v, _, found, err := st.Get(ctx, k)
if err != nil {
return nil, err
}
var p Proposal
if !found || json.Unmarshal(v, &p) != nil {
continue
}
out = append(out, ProposalRow{Key: k, Verb: p.Verb, Module: p.Module, State: p.State, Ask: p.Ask, Caller: p.Caller, At: p.At,
Outcome: p.Outcome, Ended: p.Ended, Lines: p.Lines, Digest: p.Digest})
}
return out, nil
}
func (r *Registry) load(ctx context.Context, key string) (Proposal, uint64, bool, error) {
v, rev, found, err := r.Proposals.Get(ctx, key)
if err != nil || !found {
return Proposal{}, 0, found, err
}
var p Proposal
if err := json.Unmarshal(v, &p); err != nil {
return p, rev, true, fmt.Errorf("the proposal %s cannot be read: %w", key, err)
}
return p, rev, true, nil
}
// end writes how a key ended, when it is still the open one of this ask.
func (r *Registry) end(ctx context.Context, key, ask, state, outcome string) error {
return write(ctx, r.Proposals, key, func(old []byte, found bool) ([]byte, bool, error) {
var p Proposal
if !found || json.Unmarshal(old, &p) != nil || p.Ask != ask || p.State != ProposalOpen {
return nil, false, nil
}
p.State, p.Outcome, p.Ended = state, outcome, r.now()
data, err := json.Marshal(p)
return data, false, err
})
}
// Drop is indexer_proposal_drop (class act): it cancels the key's open ask and ends the key dropped.
func (r *Registry) Drop(ctx context.Context, key, why, by string) (string, error) {
if strings.TrimSpace(why) == "" {
return "", errors.New("say why (why): a hand act is said in the hand-act log")
}
r.mu.Lock()
defer r.mu.Unlock()
p, _, found, err := r.load(ctx, key)
switch {
case err != nil:
return "", err
case !found:
return "", fmt.Errorf("no proposal %q is kept; indexer_proposals lists the keys", key)
case p.State != ProposalOpen:
return "", fmt.Errorf("the proposal %q ended %s, and has no open ask to drop", key, p.State)
}
if err := r.Cancel(p.Ask); err != nil {
return "", fmt.Errorf("the ask %s could not be cancelled: %w", p.Ask, err)
}
if err := r.end(ctx, key, p.Ask, ProposalDropped, fmt.Sprintf("dropped by %s: %s", by, why)); err != nil {
return "", err
}
if key == SeedKey && r.SeedWaits != nil {
r.SeedWaits(p.Ask, true)
}
r.logf("the proposal %s is dropped by %s (%s); its ask %s is cancelled", key, by, why, p.Ask)
return "the ask " + p.Ask + " is cancelled and the proposal " + key + " is dropped", nil
}
// Heard takes a warrant the router said for the indexer (§667 Heard). It is applied only when its ask's id is a key's current
// ask, its digest the key's, and when the router's own record of the ask holds it with Warrant.For, never the event's body
// alone. A warrant of a replaced ask writes nothing.
func (r *Registry) Heard(ctx context.Context, w oc.Warrant) error {
r.mu.Lock()
defer r.mu.Unlock()
key, p, err := r.findAsk(ctx, w.Ask)
if err != nil {
return err
}
if p == nil || p.State != ProposalOpen {
r.logf("the warrant of the ask %s (%s) matches no open proposal, so it writes nothing", w.Ask, w.Outcome)
return nil
}
if w.Outcome != occ.OutcomeChosen {
state := ProposalRefused
switch w.Outcome {
case oc.WarrantOutcomeExpired:
state = ProposalExpired
case oc.WarrantOutcomeCancelled:
state = ProposalDropped
case oc.WarrantOutcomeReplaced:
r.logf("the warrant of the ask %s says it was replaced by a newer ask of %s", w.Ask, key)
return nil
}
return r.endKey(ctx, key, *p, state, "the ask "+w.Outcome+": "+w.Words)
}
// The router's own record of the ask holds the warrant (ADR 0286 §26, §27, §36).
state, rec, err := r.Where(p.Ask)
if err != nil {
return fmt.Errorf("the router's record of %s cannot be read, so its warrant is left unapplied: %w", p.Ask, err)
}
if rec == nil {
return fmt.Errorf("the router's record of %s (%q) holds no warrant, so the warrant heard is left unapplied", p.Ask, state)
}
if rec.Ask != w.Ask || rec.Option != w.Option || rec.AskDigest != w.AskDigest || rec.Asker != w.Asker {
return r.endKey(ctx, key, *p, ProposalRefused, "the warrant heard is not the one the router's record holds")
}
o, err := occ.For(*rec, Module, p.composeAsk(key))
if err == nil {
err = occ.Performs(o, p.act())
}
if err != nil {
return r.endKey(ctx, key, *p, ProposalRefused, err.Error())
}
if o.ID == optionDecline {
return r.endKey(ctx, key, *p, ProposalDeclined, "declined: "+occ.Says(*rec))
}
if p.Verb == verbSeed {
return r.applySeed(ctx, key, *p, *rec)
}
return r.apply(ctx, key, *p, *rec)
}
func (r *Registry) endKey(ctx context.Context, key string, p Proposal, state, outcome string) error {
if key == SeedKey && state != ProposalApplied && r.SeedWaits != nil {
state = ProposalSeedWaits
r.SeedWaits(p.Ask, true)
}
if err := r.end(ctx, key, p.Ask, state, outcome); err != nil {
return err
}
r.logf("the proposal %s ended %s: %s", key, state, outcome)
return nil
}
// findAsk is the key whose current ask has the id.
func (r *Registry) findAsk(ctx context.Context, ask string) (string, *Proposal, error) {
all, err := r.Proposals.Keys(ctx)
if err != nil {
return "", nil, err
}
for _, k := range all {
p, _, found, err := r.load(ctx, k)
if err != nil || !found {
continue
}
if p.Ask == ask {
return k, &p, nil
}
}
return "", nil, nil
}
// apply writes a warranted change: its history version first (kept once under <module>.<version>), then the registration by
// compare-and-set, then the event; the key ends applied. A change composed against a registration that moved since is refused.
func (r *Registry) apply(ctx context.Context, key string, p Proposal, w oc.Warrant) error {
cur, found, err := r.Registration(ctx, p.Module)
if err != nil {
return err
}
if cur.Version != p.Was || (!found && p.Was != 0) {
return r.endKey(ctx, key, p, ProposalRefused, fmt.Sprintf("the registration of %s moved to version %d since the change was asked at version %d; ask it again", p.Module, cur.Version, p.Was))
}
next, _, err := compose(p.Verb, cur, found, p.Change)
if err != nil {
return r.endKey(ctx, key, p, ProposalRefused, err.Error())
}
return r.write(ctx, key, p, next, w)
}
func (r *Registry) write(ctx context.Context, key string, p Proposal, next meshcatalog.Registration, w oc.Warrant) error {
at := r.now()
next.V, next.Version, next.Warrant, next.By, next.At = 1, p.Was+1, w.Ask, p.Caller, &at
if r.History != nil {
if err := r.History(ctx, next); err != nil {
return fmt.Errorf("the version %d of %s is not kept on %s, so it is not written: %w", next.Version, next.Module, RegistrationHistoryStream, err)
}
}
data, err := json.Marshal(next)
if err != nil {
return err
}
err = write(ctx, r.Registrations, RegistrationKey(next.Module), func(old []byte, found bool) ([]byte, bool, error) {
var cur meshcatalog.Registration
if found && json.Unmarshal(old, &cur) == nil && cur.Version != p.Was {
return nil, false, fmt.Errorf("the registration of %s is at version %d, not %d", next.Module, cur.Version, p.Was)
}
return data, false, nil
})
if err != nil {
return err
}
if r.Say != nil {
if err := r.Say(ctx, meshcatalog.RegistrationChangedEvent{V: 1, Module: next.Module, Version: next.Version, Moved: p.Lines, By: p.Caller, Warrant: w.Ask, At: &at}); err != nil {
r.logf("registration-changed of %s %d is not said: %v", next.Module, next.Version, err)
}
}
return r.endKey(ctx, key, p, ProposalApplied, fmt.Sprintf("applied as version %d of %s on %s", next.Version, next.Module, occ.Says(w)))
}
// Restart reads each open key at the indexer's start (§667 A restart): a decided ask is applied or ended, an open one waited on,
// and one the router no longer holds ended lost, said.
func (r *Registry) Restart(ctx context.Context) error {
all, err := r.Proposals.Keys(ctx)
if err != nil {
return err
}
slices.Sort(all)
var errs []error
for _, k := range all {
r.mu.Lock()
p, _, found, err := r.load(ctx, k)
r.mu.Unlock()
if err != nil || !found || p.State != ProposalOpen {
continue
}
state, w, err := r.Where(p.Ask)
switch {
case err != nil:
errs = append(errs, fmt.Errorf("%s: %w", k, err))
case state == "" && w == nil:
r.mu.Lock()
err := r.endKey(ctx, k, p, ProposalLost, "the router no longer holds the ask "+p.Ask)
r.mu.Unlock()
if err != nil {
errs = append(errs, err)
}
case w != nil:
if err := r.Heard(ctx, *w); err != nil {
errs = append(errs, err)
}
default:
r.logf("the proposal %s waits for the ask %s (%s)", k, p.Ask, state)
}
}
return errors.Join(errs...)
}
// RegistryAct is one act of the registrations asked through the takeins state (the way every tool of the indexer asks it):
// the verb (register, policy, watch, unwatch, registrations, drop or seed), its input as JSON, and for a drop the key.
type RegistryAct struct {
Verb string `json:"verb"`
Input json.RawMessage `json:"input,omitempty"`
Key string `json:"key,omitempty"`
Again bool `json:"again,omitempty"`
}
// The verbs of a RegistryAct that are no change of a registration.
const (
ActRegistrations = "registrations"
ActDrop = "drop"
ActSeed = "seed"
)
// NewRegistryAsk is the value and the key in takeins an act of the registrations is written under.
func NewRegistryAsk(act RegistryAct, why, by string, at time.Time) (string, []byte, error) {
if act.Verb == "" {
return "", nil, errors.New("name the act")
}
data, err := json.Marshal(TakeInAsk{V: 1, Registry: &act, Why: why, By: by, At: at.UTC()})
return "registry." + act.Verb + "." + at.UTC().Format("20060102T150405.000000000"), data, err
}
// Act answers one RegistryAct: the verb's answer as JSON in Registry, or why it could not be answered. seed is what the seed is
// composed from (the trunk's modules and what their build seats answered steps with).
func (r *Registry) Act(ctx context.Context, a TakeInAsk, seed func(context.Context) ([]SeedModule, error)) TakenIn {
out := TakenIn{Module: "registry"}
act := a.Registry
if act == nil {
out.Answer = "no registry act"
return out
}
var (
answer any
err error
)
switch act.Verb {
case verbRegister, verbPolicy, verbWatch, verbUnwatch:
answer, err = r.Propose(ctx, act.Verb, a.By, act.Input)
case ActRegistrations:
var in meshcatalog.RegistrationsInput
if len(act.Input) > 0 {
err = json.Unmarshal(act.Input, &in)
}
if err == nil {
answer, err = r.RegistrationsAnswer(ctx, in)
}
case ActDrop:
var said string
said, err = r.Drop(ctx, act.Key, a.Why, a.By)
answer = map[string]string{"answer": said}
case ActSeed:
var modules []SeedModule
if seed == nil {
err = errors.New("the seed is not composed by this indexer")
} else if modules, err = seed(ctx); err == nil {
var ask string
ask, err = r.ProposeSeed(ctx, modules, act.Again, a.By)
answer = map[string]string{"ask": ask}
}
default:
err = fmt.Errorf("%q is no act of the registrations", act.Verb)
}
if err != nil {
out.Answer = err.Error()
return out
}
out.Stored = true
if out.Registry, err = json.Marshal(answer); err != nil {
out.Stored, out.Answer = false, err.Error()
return out
}
out.Answer = act.Verb + " answered"
return out
}
// RegistryAct answers a registrations' act asked through takeins (design 47 G §3, §4, §8, §667).
func (ix *Indexer) RegistryAct(ctx context.Context, a TakeInAsk) TakenIn {
if ix.Registry == nil {
return TakenIn{Module: "registry", Answer: "this indexer keeps no registrations: their states are not bound yet"}
}
return ix.Registry.Act(ctx, a, ix.SeedModules)
}
@@ -0,0 +1,294 @@
package index
import (
"context"
"encoding/json"
"errors"
"strings"
"testing"
"time"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
gitseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/git"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// regRig is a Registry on memory states with a fake router: asks it was given, the cancelled ones, the history and the events.
type regRig struct {
r *Registry
asks map[string]oc.Ask
cancelled []string
history []meshcatalog.Registration
events []meshcatalog.RegistrationChangedEvent
now time.Time
unprotect map[string]bool
whereErr error
chosen map[string]string // ask id -> option the router's record holds
missing map[string]bool
}
func newRegRig() *regRig {
g := &regRig{asks: map[string]oc.Ask{}, now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC), unprotect: map[string]bool{}, chosen: map[string]string{}, missing: map[string]bool{}}
g.r = &Registry{Registrations: MemoryStore(), Proposals: MemoryStore(),
History: func(_ context.Context, r meshcatalog.Registration) error {
g.history = append(g.history, r)
return nil
},
Say: func(_ context.Context, e meshcatalog.RegistrationChangedEvent) error {
g.events = append(g.events, e)
return nil
},
Branch: func(repo, branch string) (gitseat.BranchAnswer, error) {
if branch == "" {
branch = "main"
}
return gitseat.BranchAnswer{Branch: branch, Repository: repo, Protected: !g.unprotect[branch]}, nil
},
Ask: func(a oc.Ask) error { g.asks[a.ID] = a; return nil },
Cancel: func(id string) error { g.cancelled = append(g.cancelled, id); return nil },
Where: func(id string) (string, *oc.Warrant, error) {
if g.whereErr != nil {
return "", nil, g.whereErr
}
if g.missing[id] {
return "", nil, nil
}
if opt, ok := g.chosen[id]; ok {
w := g.warrant(id, opt)
return "chosen", &w, nil
}
return "open", nil, nil
},
Now: func() time.Time { return g.now }, Logf: func(string, ...any) {}}
return g
}
// warrant is the warrant of the operator's choice of option on an ask, as the router would say it.
func (g *regRig) warrant(id, option string) oc.Warrant {
a := g.asks[id]
o, _ := occ.OptionOf(a, option)
return oc.Warrant{Ask: id, Asker: Module, Outcome: occ.OutcomeChosen, Option: option, Level: o.Level, Label: o.Label,
AskDigest: occ.Digest(a), At: occ.FormatTime(g.now.Add(time.Minute)), By: &oc.Person{Who: occ.Operator, Kind: "telegram", Verified: "user id verified"}}
}
func (g *regRig) decide(t *testing.T, ask, option string) {
t.Helper()
g.chosen[ask] = option
if err := g.r.Heard(context.Background(), g.warrant(ask, option)); err != nil {
t.Fatalf("hearing the warrant of %s: %v", ask, err)
}
}
func reg(module string, steps meshcatalog.StepStatuses, branches ...string) json.RawMessage {
b, _ := json.Marshal(meshcatalog.RegisterInput{Registration: meshcatalog.Registration{Module: module, Origin: "novox/" + module, Path: "module.json", Branches: branches, Policy: meshcatalog.Policy{Steps: steps}}})
return b
}
func pol(module string, s meshcatalog.StepStatuses) json.RawMessage {
b, _ := json.Marshal(meshcatalog.PolicyInput{Module: module, Steps: s})
return b
}
func TestRegistrationsAreWrittenOnTheirWarrantOnly(t *testing.T) {
ctx := context.Background()
g := newRegRig()
// Refusals raise no ask: a code review that is required (§3), a branch named candidate or entry (§343), an unprotected branch (§5).
g.unprotect["beta"] = true
for _, c := range []struct {
name, why string
change json.RawMessage
}{
{"review", meshcatalog.RegistrationAskedRefusedReviewRequired, reg("a", meshcatalog.StepStatuses{Review: "required"}, "main")},
{"unprotected", meshcatalog.RegistrationAskedRefusedUnprotectedBranch, reg("a", meshcatalog.StepStatuses{}, "beta")},
{"candidate", meshcatalog.RegistrationAskedRefusedUnprotectedBranch, reg("a", meshcatalog.StepStatuses{}, "candidate")},
} {
if c.name == "candidate" {
// the name is refused whatever the forge says of it
c.why = meshcatalog.RegistrationAskedRefusedUnprotectedBranch
}
got, err := g.r.Propose(ctx, "register", "session@node", c.change)
if err != nil || got.Refused == "" || got.Ask != "" || got.Words == "" {
t.Fatalf("%s: answer %+v, %v: want a refusal naming why and no ask", c.name, got, err)
}
}
if len(g.asks) != 0 {
t.Fatalf("a refusal raised %d asks", len(g.asks))
}
// A registration with no branch takes the default branch through the seat git (§342) and asks, naming each value and carrying the digest.
got, err := g.r.Propose(ctx, "register", "session@node", reg("a", meshcatalog.StepStatuses{Unit: "required", Lint: "bypass"}))
if err != nil || got.Ask == "" || len(got.Lines) < 3 {
t.Fatalf("register: %+v, %v", got, err)
}
ask := g.asks[got.Ask]
if !strings.Contains(ask.Explanation, "watching main") || !strings.Contains(ask.Explanation, "sha256:") || ask.About == "" {
t.Fatalf("the ask does not name the values and the digest: %+v", ask)
}
if all, _ := g.r.RegistrationsAnswer(ctx, meshcatalog.RegistrationsInput{}); len(all.Registrations) != 0 {
t.Fatalf("a registration was written before its warrant: %+v", all)
}
g.decide(t, got.Ask, "approve")
one, _ := g.r.RegistrationsAnswer(ctx, meshcatalog.RegistrationsInput{Module: "a"})
if len(one.Registrations) != 1 || one.Registrations[0].Version != 1 || one.Registrations[0].Warrant != got.Ask || one.Registrations[0].By != "session@node" {
t.Fatalf("after the warrant: %+v", one)
}
if len(g.history) != 1 || len(g.events) != 1 || g.events[0].Version != 1 || g.events[0].Warrant != got.Ask {
t.Fatalf("history %d, events %+v", len(g.history), g.events)
}
// A policy from required to bypass asks, showing that line; a decline writes nothing and ends the key declined.
p, err := g.r.Propose(ctx, "policy", "s@n", pol("a", meshcatalog.StepStatuses{Unit: "bypass"}))
if err != nil || len(p.Lines) != 1 || p.Lines[0] != "a unit tests: required -> bypass" {
t.Fatalf("policy: %+v, %v", p, err)
}
g.decide(t, p.Ask, "decline")
if now, _, _ := g.r.Registration(ctx, "a"); now.Version != 1 {
t.Fatalf("a declined change was written: %+v", now)
}
rows, _ := ListProposals(ctx, g.r.Proposals)
if len(rows) != 2 || rows[1].State != ProposalDeclined && rows[0].State != ProposalDeclined {
t.Fatalf("proposals: %+v", rows)
}
// Two changes of one key leave one open ask; the first ask's late warrant writes nothing.
g.now = g.now.Add(time.Hour)
first, _ := g.r.Propose(ctx, "policy", "s@n", pol("a", meshcatalog.StepStatuses{Unit: "bypass"}))
g.now = g.now.Add(time.Hour)
second, _ := g.r.Propose(ctx, "policy", "s@n", pol("a", meshcatalog.StepStatuses{Unit: "optional"}))
if first.Ask == second.Ask {
t.Fatal("a newer change of the key kept the older ask")
}
g.chosen[first.Ask] = "approve"
_ = g.r.Heard(ctx, g.warrant(first.Ask, "approve"))
if now, _, _ := g.r.Registration(ctx, "a"); now.Version != 1 {
t.Fatalf("the replaced ask's warrant wrote: %+v", now)
}
// Where the router's record cannot be read the warrant is left unapplied, said as an error.
g.whereErr = errors.New("no runtime")
if err := g.r.Heard(ctx, g.warrant(second.Ask, "approve")); err == nil {
t.Fatal("a warrant was applied with the router's record unread")
}
g.whereErr = nil
g.decide(t, second.Ask, "approve")
if now, _, _ := g.r.Registration(ctx, "a"); now.Version != 2 || now.Policy.Steps.Unit != "optional" {
t.Fatalf("the second change: %+v", now)
}
// watch adds a branch only when protected; unwatch removes it; each on its warrant.
w, _ := g.r.Propose(ctx, "watch", "s@n", mustJSON(meshcatalog.WatchInput{Module: "a", Branch: "beta"}))
if w.Refused != meshcatalog.RegistrationAskedRefusedUnprotectedBranch {
t.Fatalf("watch of an unprotected branch: %+v", w)
}
delete(g.unprotect, "beta")
w, _ = g.r.Propose(ctx, "watch", "s@n", mustJSON(meshcatalog.WatchInput{Module: "a", Branch: "beta"}))
g.decide(t, w.Ask, "approve")
u, _ := g.r.Propose(ctx, "unwatch", "s@n", mustJSON(meshcatalog.WatchInput{Module: "a", Branch: "beta"}))
g.decide(t, u.Ask, "approve")
if now, _, _ := g.r.Registration(ctx, "a"); now.Version != 4 || len(now.Branches) != 1 {
t.Fatalf("watch and unwatch: %+v", now)
}
if un, _ := g.r.Propose(ctx, "policy", "s@n", pol("zz", meshcatalog.StepStatuses{Unit: "bypass"})); un.Refused != meshcatalog.RegistrationAskedRefusedUnknownModule {
t.Fatalf("a policy of an unregistered module: %+v", un)
}
}
func mustJSON(v any) json.RawMessage { b, _ := json.Marshal(v); return b }
func TestProposalsSurviveARestartAndAreDropped(t *testing.T) {
ctx := context.Background()
g := newRegRig()
a, _ := g.r.Propose(ctx, "register", "s@n", reg("a", meshcatalog.StepStatuses{}, "main"))
b, _ := g.r.Propose(ctx, "register", "s@n", reg("b", meshcatalog.StepStatuses{}, "main"))
c, _ := g.r.Propose(ctx, "register", "s@n", reg("c", meshcatalog.StepStatuses{}, "main"))
// While the indexer was down the operator approved a and the router lost c; b is still open.
g.chosen[a.Ask] = "approve"
g.missing[c.Ask] = true
if err := g.r.Restart(ctx); err != nil {
t.Fatal(err)
}
rows, _ := ListProposals(ctx, g.r.Proposals)
states := map[string]string{}
for _, row := range rows {
states[row.Module] = row.State
}
if states["a"] != ProposalApplied || states["b"] != ProposalOpen || states["c"] != ProposalLost {
t.Fatalf("after a restart: %v", states)
}
// indexer_proposal_drop cancels the open ask and ends the key dropped; it asks why.
if _, err := g.r.Drop(ctx, ProposalKey("register", "b", ""), "", "s@n"); err == nil {
t.Fatal("a drop with no why was taken")
}
if _, err := g.r.Drop(ctx, ProposalKey("register", "b", ""), "not wanted", "s@n"); err != nil {
t.Fatal(err)
}
if len(g.cancelled) != 1 || g.cancelled[0] != b.Ask {
t.Fatalf("cancelled %v", g.cancelled)
}
rows, _ = ListProposals(ctx, g.r.Proposals)
for _, row := range rows {
if row.Module == "b" && row.State != ProposalDropped {
t.Fatalf("a dropped key: %+v", row)
}
}
}
func TestSeedIsOneAskAndWaitsForEveryBuildSeat(t *testing.T) {
ctx := context.Background()
g := newRegRig()
g.unprotect["main"] = false
mods := []SeedModule{{Module: "a", Origin: "novox/a", Path: "module.json", Artifacts: []SeedArtifact{{Name: "program", Seat: "node-build-go", Answered: false}}}}
if _, err := g.r.ProposeSeed(ctx, mods, false, "indexer"); err == nil || !strings.Contains(err.Error(), "node-build-go") {
t.Fatalf("an unanswered build seat did not hold the seed: %v", err)
}
mods[0].Artifacts[0].Answered, mods[0].Artifacts[0].Steps = true, []string{"unit"}
waits := ""
g.r.SeedWaits = func(ask string, waiting bool) {
if waiting {
waits = ask
} else {
waits = ""
}
}
ask, err := g.r.ProposeSeed(ctx, mods, false, "indexer")
if err != nil || ask == "" {
t.Fatalf("seed: %q, %v", ask, err)
}
// A step the seat answers none for is bypass; the statuses the check judged are the rest.
g.decide(t, ask, "decline")
if waits != ask {
t.Fatalf("a declined seed does not keep seed-waits: %q", waits)
}
if _, err := g.r.ProposeSeed(ctx, mods, false, "indexer"); err == nil {
t.Fatal("the seed was asked again with no indexer_seed_again")
}
g.now = g.now.Add(time.Hour)
ask, err = g.r.ProposeSeed(ctx, mods, true, "s@n")
if err != nil {
t.Fatal(err)
}
g.decide(t, ask, "approve")
got, _, _ := g.r.Registration(ctx, "a")
if got.Version != 1 || got.Policy.Steps.Review != "optional" || got.Policy.Steps.Approval != "bypass" || got.Policy.Artifacts["program"].Lint != "bypass" || got.Policy.Artifacts["program"].Integration != "bypass" || got.Policy.Artifacts["program"].Unit != "" {
t.Fatalf("the seeded registration: %+v", got)
}
if waits != "" {
t.Fatalf("seed-waits is not cleared by the warrant: %q", waits)
}
// An unprotected default branch is left out and named.
g2 := newRegRig()
g2.unprotect["main"] = true
seed, err := g2.r.ComposeSeed(mods)
if err != nil || len(seed.Registrations) != 0 || len(seed.Left) != 1 || !strings.Contains(seed.Left[0], "not protected") {
t.Fatalf("seed with an unprotected default branch: %+v, %v", seed, err)
}
}
func TestCanonicalIsRFC8785(t *testing.T) {
got, err := Canonical(map[string]any{"b": []any{1, "x
"}, "a": map[string]any{"z": true, "y": nil}, "é": "<&>"})
if err != nil {
t.Fatal(err)
}
if want := "{\"a\":{\"y\":null,\"z\":true},\"b\":[1,\"x
\"],\"é\":\"<&>\"}"; string(got) != want {
t.Fatalf("canonical = %s, want %s", got, want)
}
if _, err := Canonical(map[string]any{"n": 1.5}); err == nil {
t.Fatal("a fraction has no canonical form here")
}
}
@@ -0,0 +1,119 @@
package index
import (
"context"
"encoding/json"
"errors"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// RegistrationTools are the seat mesh-catalog's verbs registrations, register, policy, watch and unwatch (design 47 G §3, §4), and
// the indexer's own tools of the proposals (§667) and the seed (§8), served in the tools bundle. Every act is asked of the
// indexer through takeins, as the indexer's other tools are (ask writes the value and waits for the answer); the proposals
// are read from the state as the bundle reads it. record says a hand act in the hand-act log.
func RegistrationTools(proposals Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(cause, verb, what, why string) error) []stdio.Tool {
act := func(verb string, withInput bool) func(map[string]any) (any, error) {
return func(args map[string]any) (any, error) {
var input json.RawMessage
if withInput {
clean := map[string]any{}
for k, v := range args {
if k != "caller" && !strings.HasPrefix(k, "_") {
clean[k] = v
}
}
var err error
if input, err = json.Marshal(clean); err != nil {
return nil, err
}
}
why, _ := args["why"].(string)
key, value, err := NewRegistryAsk(RegistryAct{Verb: verb, Input: input}, why, stdio.CallerOf(args), now())
if err != nil {
return nil, err
}
return answerOf(ask(key, value))
}
}
return []stdio.Tool{
{Name: meshcatalog.Seat + "." + meshcatalog.VerbRegistrations, Run: act(ActRegistrations, true)},
{Name: meshcatalog.Seat + "." + meshcatalog.VerbRegister, Run: act(verbRegister, true)},
{Name: meshcatalog.Seat + "." + meshcatalog.VerbPolicy, Run: act(verbPolicy, true)},
{Name: meshcatalog.Seat + "." + meshcatalog.VerbWatch, Run: act(verbWatch, true)},
{Name: meshcatalog.Seat + "." + meshcatalog.VerbUnwatch, Run: act(verbUnwatch, true)},
{Name: ProposalsTool, Description: "The changes of the registrations that wait for the operator's warrant and those that ended (design 47 G §667): " +
"for each key (<verb>.<module>, with the branch for watch and unwatch, and seed) its state (open, applied, declined, " +
"expired, refused, dropped, lost or seed-waits), its ask, caller, the lines it moves, its digest and its outcome. An ended key stays until the next change of the same key replaces it.",
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) {
rows, err := ListProposals(context.Background(), proposals)
return map[string]any{"proposals": rows}, err
}},
{Name: ProposalDropTool, Description: "Cancel the open ask of one key of the proposals and end the key dropped (design 47 G §667). A hand act: who and why " +
"are said in the hand-act log. indexer_proposals names the keys.",
Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{
"key": map[string]any{"type": "string", "description": "the proposal's key, as indexer_proposals names it"},
"why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}},
Run: func(args map[string]any) (any, error) {
key, _ := args["key"].(string)
why, _ := args["why"].(string)
if strings.TrimSpace(key) == "" || strings.TrimSpace(why) == "" {
return nil, errors.New("name the key (key) and say why (why): a hand act is said in the hand-act log")
}
k, value, err := NewRegistryAsk(RegistryAct{Verb: ActDrop, Key: strings.TrimSpace(key)}, why, stdio.CallerOf(args), now())
if err != nil {
return nil, err
}
out, err := answerOf(ask(k, value))
if err != nil {
return out, err
}
res := map[string]any{"result": out}
if rerr := record(ProposalDropTool, "proposal-drop", "dropped the proposal "+key, why); rerr != nil {
res["hand-act"] = "not recorded: " + rerr.Error()
}
return res, nil
}},
{Name: SeedAgainTool, Description: "Raise the ask of the first registrations again from the seed's key (design 47 G §8), after a Decline, an expiry or a drop " +
"ended it and while the indexer holds no registration. A hand act: who and why are said in the hand-act log.",
Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{
"why": map[string]any{"type": "string", "description": "why it is asked again, in plain words"}}},
Run: func(args map[string]any) (any, error) {
why, _ := args["why"].(string)
if strings.TrimSpace(why) == "" {
return nil, errors.New("say why (why): a hand act is said in the hand-act log")
}
k, value, err := NewRegistryAsk(RegistryAct{Verb: ActSeed, Again: true}, why, stdio.CallerOf(args), now())
if err != nil {
return nil, err
}
out, err := answerOf(ask(k, value))
if err != nil {
return out, err
}
res := map[string]any{"result": out}
if rerr := record(SeedAgainTool, "seed-again", "asked the first registrations again", why); rerr != nil {
res["hand-act"] = "not recorded: " + rerr.Error()
}
return res, nil
}},
}
}
// answerOf is the answer of an act the indexer answered: the verb's own answer, or the words that say why not.
func answerOf(a TakeInAsk, err error) (any, error) {
if err != nil {
return nil, err
}
if a.Result == nil {
return nil, errors.New("the indexer gave no answer")
}
if !a.Result.Stored {
return nil, errors.New(a.Result.Answer)
}
return a.Result.Registry, nil
}
@@ -0,0 +1,171 @@
package index
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"strings"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// SeedModule is one module the trunk revision holds, as the seed reads it (design 47 G §8): its origin and the path of its
// manifest, and for each artifact whether its build seat answered `steps` and with which steps.
type SeedModule struct {
Module string
Origin string
Path string
Artifacts []SeedArtifact
}
// SeedArtifact is an artifact of a module and what its build seat answered `steps` with: Answered false is a seat that has not
// answered, which holds the seed (never read as none), and Steps empty with Answered true is none, so every step the
// artifact's seat detects none for is bypass.
type SeedArtifact struct {
Name string
Seat string
Answered bool
Steps []string
}
// SeedProposal is the first registrations composed from what the indexer took in (§8): the registrations proposed, the
// modules left out because their default branch is not protected, and the build seats still unanswered.
type SeedProposal struct {
Registrations []meshcatalog.Registration
Left []string
Unanswered []string
}
// ComposeSeed proposes one registration for every module its trunk revision holds (§8): the origin and path of each, its
// watched branch the repository's default branch read through the seat git, and the statuses the check judged until then:
// lint, unit tests, integration tests and composition required, code review optional, approval bypass; each step an
// artifact's build seat answers steps with none is bypass, so no module is held by a required step it has nothing for. A
// module whose default branch the forge does not protect is left out and named. A build seat that has not answered holds
// the seed.
func (r *Registry) ComposeSeed(modules []SeedModule) (SeedProposal, error) {
var out SeedProposal
slices.SortFunc(modules, func(a, b SeedModule) int { return strings.Compare(a.Module, b.Module) })
for _, m := range modules {
for _, a := range m.Artifacts {
if !a.Answered {
out.Unanswered = append(out.Unanswered, a.Seat)
}
}
}
slices.Sort(out.Unanswered)
out.Unanswered = slices.Compact(out.Unanswered)
if len(out.Unanswered) > 0 {
return out, nil
}
for _, m := range modules {
b, err := r.Branch(m.Origin, "")
if err != nil {
return out, fmt.Errorf("the seat git does not answer the default branch of %s: %w", m.Origin, err)
}
if b.Error != "" || b.Branch == "" {
return out, fmt.Errorf("the default branch of %s is not read: %s", m.Origin, b.Error)
}
if p, err := r.Branch(m.Origin, b.Branch); err != nil || !p.Protected || p.Error != "" {
out.Left = append(out.Left, fmt.Sprintf("%s (the branch %s of %s is not protected)", m.Module, b.Branch, m.Origin))
continue
}
reg := meshcatalog.Registration{Module: m.Module, Origin: m.Origin, Path: m.Path, Branches: []string{b.Branch},
Policy: meshcatalog.Policy{Steps: meshcatalog.StepStatuses{Lint: "required", Unit: "required", Integration: "required",
Composition: "required", Review: "optional", Approval: "bypass"}}}
for _, a := range m.Artifacts {
var s meshcatalog.StepStatuses
if !slices.Contains(a.Steps, "lint") {
s.Lint = "bypass"
}
if !slices.Contains(a.Steps, "unit") {
s.Unit = "bypass"
}
if !slices.Contains(a.Steps, "integration") {
s.Integration = "bypass"
}
if s != (meshcatalog.StepStatuses{}) {
if reg.Policy.Artifacts == nil {
reg.Policy.Artifacts = map[string]meshcatalog.StepStatuses{}
}
reg.Policy.Artifacts[a.Name] = s
}
}
out.Registrations = append(out.Registrations, reg)
}
return out, nil
}
// ProposeSeed raises the first registrations as one ask (§8), when the indexer holds no registration and the seed's key is not
// open or applied; again (indexer_seed_again) it raises the ask once more from a key a Decline, an expiry or a drop ended.
// It answers the ask's id, or why none was raised.
func (r *Registry) ProposeSeed(ctx context.Context, modules []SeedModule, again bool, by string) (string, error) {
r.mu.Lock()
defer r.mu.Unlock()
held, err := r.Registrations.Keys(ctx)
if err != nil {
return "", err
}
if len(held) > 0 {
return "", errors.New("the indexer holds registrations already: the seed is for the first start only")
}
if old, _, found, err := r.load(ctx, SeedKey); err != nil {
return "", err
} else if found && old.State == ProposalOpen {
return old.Ask, nil
} else if found && !again {
return "", fmt.Errorf("the seed ended %s: indexer_seed_again asks it again", old.State)
}
seed, err := r.ComposeSeed(modules)
if err != nil {
return "", err
}
if len(seed.Unanswered) > 0 {
return "", fmt.Errorf("the build seats %v have not answered steps: the seed waits for them (call-unanswered after five minutes)", seed.Unanswered)
}
if len(seed.Registrations) == 0 {
return "", errors.New("no module of the trunk revision can be registered: " + strings.Join(seed.Left, "; "))
}
change, err := json.Marshal(seed.Registrations)
if err != nil {
return "", err
}
p := Proposal{V: 1, Verb: verbSeed, Change: change, Seed: seed.Registrations, Caller: by, At: r.now(), State: ProposalOpen}
for _, g := range seed.Registrations {
p.Lines = append(p.Lines, fmt.Sprintf("%s: registered from %s at %s, watching %s", g.Module, g.Origin, g.Path, strings.Join(g.Branches, ", ")))
}
for _, l := range seed.Left {
p.Lines = append(p.Lines, "left out: "+l)
}
if p.Digest, err = DigestOf(map[string]any{"verb": verbSeed, "change": json.RawMessage(change)}); err != nil {
return "", err
}
if err := r.keepAndAsk(ctx, SeedKey, &p); err != nil {
return "", err
}
if r.SeedWaits != nil {
r.SeedWaits(p.Ask, false)
}
return p.Ask, nil
}
// applySeed writes every registration the seed proposes, each as version 1 on the warrant, and ends the key applied.
func (r *Registry) applySeed(ctx context.Context, key string, p Proposal, w oc.Warrant) error {
if held, err := r.Registrations.Keys(ctx); err != nil || len(held) > 0 {
return r.endKey(ctx, key, p, ProposalRefused, "the indexer holds registrations already, so the seed writes none")
}
for _, reg := range p.Seed {
q := p
q.Verb, q.Module, q.Was = verbRegister, reg.Module, 0
q.Lines = []string{fmt.Sprintf("%s: registered by the first registrations", reg.Module)}
if err := r.write(ctx, "", q, reg, w); err != nil {
return err
}
}
if r.SeedWaits != nil {
r.SeedWaits(p.Ask, false)
}
return r.end(ctx, key, p.Ask, ProposalApplied, fmt.Sprintf("applied: %d registrations on %s", len(p.Seed), w.Ask))
}
@@ -67,13 +67,16 @@ type TakeInAsk struct {
// Recompute asks the catalog graph to be computed again now and compared with the newest kept revision (indexer_recompute).
Recompute bool `json:"recompute,omitempty"`
// CandidateGroup and CandidateHeads, with Drop, ask that candidate's catalog half to be dropped (indexer_drop_candidate_graph).
CandidateGroup string `json:"candidate-group,omitempty"`
CandidateHeads string `json:"candidate-heads,omitempty"`
Drop bool `json:"drop,omitempty"`
By string `json:"by,omitempty"`
At time.Time `json:"at"`
Result *TakenIn `json:"result,omitempty"`
Answered time.Time `json:"answered,omitzero"`
CandidateGroup string `json:"candidate-group,omitempty"`
CandidateHeads string `json:"candidate-heads,omitempty"`
Drop bool `json:"drop,omitempty"`
// Registry asks one of the registrations' acts (registrations.go): a trusted change of one, the verb registrations,
// indexer_proposal_drop or indexer_seed_again (design 47 G §3, §4, §8, §667).
Registry *RegistryAct `json:"registry,omitempty"`
By string `json:"by,omitempty"`
At time.Time `json:"at"`
Result *TakenIn `json:"result,omitempty"`
Answered time.Time `json:"answered,omitzero"`
}
// Done says the ask is answered.
@@ -88,7 +91,7 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) {
if a.V != 1 {
return a, fmt.Errorf("a takeins value of version %d, not 1", a.V)
}
if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" {
if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.Registry == nil {
return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try")
}
return a, nil
@@ -174,6 +174,18 @@ func (ix *Indexer) judgeOverHeld(ctx context.Context, fresh map[string]bool, ove
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)
}
// A package's name belongs to one module (design 47 G §491, §676): the trunk's manifests only, since a candidate's
// revision is validated by the validator built from its own heads (§16).
if over == nil {
if refused == nil {
refused = map[string][]string{}
}
for module, p := range judge.PackageNameProblems(ix.packageEntries(ctx, held, raws)) {
if _, waiting := waits[module]; !waiting {
refused[module] = append(refused[module], p...)
}
}
}
out.Refused = refused
// A manifest with a field this judge does not know waits for a judge that may know it (clause 3 of the record on a
// release's Approve; ADR 0578 §1): it is refused only when this program is the newest build of the indexer the trunk
@@ -371,8 +371,8 @@ func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) {
}
}
if !slices.Equal(judge.Foundation, []string{"nats", "mesh-grants", "mesh-delivery-indexer", "mesh-delivery-assigner", "node-tools",
"mesh-delivery-builder", "mesh-delivery-checker", "mesh-config", "mesh-vault", "mesh-delivery-pipeline"}) {
t.Fatalf("the foundation is %v, want exactly the nine of ADR 0619 and the pipeline runner (ADR 0666 §4)", judge.Foundation)
"mesh-delivery-builder", "mesh-delivery-checker", "mesh-config", "mesh-vault", "mesh-delivery-pipeline", "gitea", "distribution"}) {
t.Fatalf("the foundation is %v, want exactly the nine of ADR 0619, the pipeline runner (ADR 0666 §4) and the forge and image registry modules (design 47 G §584)", judge.Foundation)
}
late := hold("late", `{"module":"late","version":"1.0","logs":[{"name":"l"}]}`)
grants := hold("mesh-grants", `{"module":"mesh-grants","version":"2.0","logs":[{"name":"l"},{"name":"m"}]}`)
@@ -553,3 +553,48 @@ type unreadable struct{ Store }
func (unreadable) Get(context.Context, string) ([]byte, uint64, bool, error) {
return nil, 0, false, errors.New("the bus did not answer")
}
// Design 47 G §491, §676: two modules whose entries carry one Go module path are both refused by the trunk's judge pass,
// each naming both; a module with its own name is accepted.
func TestAPackageNameBelongsToOneModule(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) }}
ix.ReadFile = func(_ context.Context, _, _, path string) ([]byte, error) {
if strings.HasSuffix(path, "go.mod") {
name := "git.example/own/" + strings.Split(path, "/")[0]
if strings.HasPrefix(path, "one/") || strings.HasPrefix(path, "two/") {
name = "git.example/shared"
}
return []byte("module " + name + "\n"), nil
}
return nil, nil
}
hold := func(module string) string {
raw := `{"module":"` + module + `","version":"1.0","build":{"artifacts":[{"name":"go","kind":"package","language":"go","paths":["go"],"series":"0.1","generated-from":"."}]}}`
digest, err := ix.keepManifest(ctx, module, raw)
if err != nil {
t.Fatal(err)
}
value, err := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: "src2:" + strings.Repeat("a", 64), Manifest: digest,
Repository: "novox/r", Path: module, Commit: "c"})
if err != nil {
t.Fatal(err)
}
ix.Modules.(*memStore).put(module, value)
return digest
}
digests := map[string]bool{hold("one"): true, hold("two"): true, hold("three"): true}
j, err := ix.JudgeAll(ctx, digests)
if err != nil {
t.Fatal(err)
}
if len(j.Refused) != 2 || j.Refused["one"] == nil || j.Refused["two"] == nil || j.Refused["three"] != nil {
t.Fatalf("refused %v, want one and two", j.Refused)
}
for _, m := range []string{"one", "two"} {
if p := strings.Join(j.Refused[m], " "); !strings.Contains(p, "one and two") || !strings.Contains(p, "git.example/shared") {
t.Errorf("%s's problems %q do not name both modules and the package", m, p)
}
}
}
@@ -160,7 +160,7 @@ func judgedByRequest(t *testing.T, shelf catalogue.Shelf, known KnownPackages) s
if known != nil {
names = append([]string{}, slices.Sorted(maps.Keys(known))...)
}
c := composer.Case{Name: "change", Manifests: map[string]composer.Manifest{}}
c := composer.Case{Name: "change", Manifests: map[string]composer.Manifest{"node-engine": {Raw: []byte(engineManifest("node-engine"))}}}
for name, m := range shelf {
raw, err := json.Marshal(m)
if err != nil {
@@ -2,6 +2,7 @@ package judge
import (
"git.novox.be/novox/mesh-sdk/go/catalogue"
"regexp"
"strings"
"testing"
)
@@ -16,16 +17,32 @@ func TestTheJudgeReadsMilestoneGsWords(t *testing.T) {
{name: "MAJOR.MINOR", raw: `{"module":"a","version":"1.4"}`},
{name: "a pre-release suffix", raw: `{"module":"a","version":"1.4-beta"}`, want: "version"},
{name: "three numbers", raw: `{"module":"a","version":"1.4.2"}`, want: "version"},
{name: "a findings state without its stream", raw: `{"module":"a","version":"1.0","state":["findings"]}`, want: "findings"},
{name: "a findings state with its stream and ask", raw: `{"module":"a","version":"1.0","state":["findings"],` +
`"streams":[{"name":"a-findings","events":["finding-written"],"retention":"history"}],"consumes":["mesh-conditions.findings-asked"]}`},
{name: "a foundation part's findings without its stream", raw: `{"module":"gitea","version":"1.0","state":["findings"]}`, want: "findings"},
{name: "a foundation part's findings with its stream and ask (§529, §584)", raw: `{"module":"gitea","version":"1.0","state":["findings"],` +
`"streams":[{"name":"gitea-findings","events":["finding-written"],"retention":"history"}],"consumes":["mesh-conditions.findings-asked"]}`},
{name: "the image registry is a foundation part too (§584)", raw: `{"module":"distribution","version":"1.0","state":["findings"],` +
`"streams":[{"name":"distribution-findings","events":["finding-written"],"retention":"history"}],"consumes":["mesh-conditions.findings-asked"]}`},
{name: "another part's old findings and stream (§529, §577)", raw: `{"module":"a","version":"1.0","state":["findings"],` +
`"streams":[{"name":"a-findings","events":["finding-written"],"retention":"history"}],"consumes":["mesh-conditions.findings-asked"]}`, want: "old sections"},
{name: "another part's findings in its requires (§529)", raw: `{"module":"a","version":"1.0","consumes":["mesh-conditions.findings-asked"],"requires":[` +
`{"name":"findings","storage":"key-value","template":"state"},` +
`{"name":"finding-events","storage":"stream","template":"events","options":{"events":["finding-written.*","finding-cleared.*","findings-stated.*"],"max-age-days":7,"max-mib":16}}]}`},
{name: "another part's findings without the finding events (§529)", raw: `{"module":"a","version":"1.0","consumes":["mesh-conditions.findings-asked"],"requires":[` +
`{"name":"findings","storage":"key-value","template":"state"}]}`, want: "finding-events"},
{name: "another part's finding events not kept 7 days (§529)", raw: `{"module":"a","version":"1.0","consumes":["mesh-conditions.findings-asked"],"requires":[` +
`{"name":"findings","storage":"key-value","template":"state"},` +
`{"name":"finding-events","storage":"stream","template":"events","options":{"events":["finding-written.*","finding-cleared.*","findings-stated.*"],"max-age-days":30,"max-mib":16}}]}`, want: "max-age-days"},
} {
raws := map[string][]byte{"a": []byte(row.raw)}
module := "a"
if m := regexp.MustCompile(`"module":"([a-z-]+)"`).FindStringSubmatch(row.raw); m != nil {
module = m[1]
}
raws := map[string][]byte{module: []byte(row.raw)}
var refused map[string][]string
WithSeats([]catalogue.Seat{{Name: "thing", Scope: catalogue.ScopeMesh, Emits: []string{"finding-written"}}}, func() {
refused, _ = Shelf(raws, nil, 0)
})
got := strings.Join(refused["a"], "\n")
got := strings.Join(refused[module], "\n")
if (row.want == "") != (got == "") || !strings.Contains(got, row.want) {
t.Errorf("%s: refused with %q, want %q", row.name, got, row.want)
}
+96 -12
View File
@@ -1,9 +1,11 @@
package judge
import (
"encoding/json"
"fmt"
"regexp"
"slices"
"strings"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
@@ -30,25 +32,107 @@ func versionProblems(m catalogue.Manifest) []string {
"and no pre-release or custom suffix exists (design 47 G §368)", m.Module, m.Version)}
}
// gWordsProblems are the problems of a manifest's words milestone G adds: the finding stream and
// ask of a module that declares a state `findings` (§387). The `*` emitter of `consumes` (§386) is judged with the
// other consumed patterns (consumePattern).
// FindingEventsEntry is the local name of the stream a part that is no foundation part asks for its findings, by the template
// `events` (design 47 G §529).
const FindingEventsEntry = "finding-events"
// findingEvents are the three events a finding stream lists (design 47 G §383, §529).
var findingEvents = []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}
// gWordsProblems are the problems of a manifest's words milestone G adds: the finding state and stream and the ask of a
// module that declares a state `findings` (§387), in the form of its part (§529, §577): a foundation part (Foundation)
// declares the state in its old `state` section and the stream `<module>-findings` in its old `streams` section; every other
// part asks both in its `requires`, the state `findings` by the template `state` and the stream `finding-events` by the
// template `events`, and the old stream is refused. The `*` emitter of `consumes` (§386) is judged with the other consumed
// patterns (consumePattern).
func gWordsProblems(m catalogue.Manifest) []string {
var problems []string
for _, s := range m.State {
if s.Name != "findings" {
continue
oldState := slices.ContainsFunc(m.State, func(s catalogue.StateDeclaration) bool { return s.Name == "findings" })
stream := FindingStreamOf(m.Module)
oldStream := slices.ContainsFunc(m.Streams, func(d catalogue.StreamDeclaration) bool { return d.Name == stream })
var asked, askedStream *catalogue.BusStateEntry
for i, e := range m.BusState {
switch {
case e.Name == "findings" && e.Template == catalogue.TemplateState:
asked = &m.BusState[i]
case e.Name == FindingEventsEntry && e.Template == catalogue.TemplateEvents:
askedStream = &m.BusState[i]
}
stream := FindingStreamOf(m.Module)
if !slices.ContainsFunc(m.Streams, func(d catalogue.StreamDeclaration) bool { return d.Name == stream }) {
}
if !oldState && asked == nil {
return nil
}
if slices.Contains(Foundation, m.Module) {
if !oldStream {
problems = append(problems, fmt.Sprintf("%s declares the state findings without the stream %q its findings are said on "+
"(design 47 G §387)", m.Module, stream))
}
if !slices.Contains(m.Consumes, FindingsAsked) {
problems = append(problems, fmt.Sprintf("%s declares the state findings without consuming %q, with which the "+
"keeper asks its findings again (design 47 G §387)", m.Module, FindingsAsked))
} else {
if oldState || oldStream {
problems = append(problems, fmt.Sprintf("%s declares its findings in its old sections, which only a foundation part does; "+
"it asks the state `findings` by the template `state` and the stream %q by the template `events` in `requires` "+
"(design 47 G §529, §577)", m.Module, FindingEventsEntry))
}
break
if askedStream == nil {
problems = append(problems, fmt.Sprintf("%s declares the state findings without the stream %q its findings are said on, "+
"asked by the template `events` in `requires` (design 47 G §529)", m.Module, FindingEventsEntry))
} else {
problems = append(problems, findingEventsProblems(m.Module, *askedStream)...)
}
}
if !slices.Contains(m.Consumes, FindingsAsked) {
problems = append(problems, fmt.Sprintf("%s declares the state findings without consuming %q, with which the "+
"keeper asks its findings again (design 47 G §387)", m.Module, FindingsAsked))
}
return problems
}
// findingEventsProblems judges the entry `finding-events` of a part that is no foundation part: it lists the three finding
// events in its option `events` and keeps them 7 days and 16 MiB (design 47 G §529).
func findingEventsProblems(module string, e catalogue.BusStateEntry) []string {
var problems []string
var listed []string
if raw, ok := e.Options["events"]; ok {
_ = json.Unmarshal(raw, &listed)
}
for _, want := range findingEvents {
if !slices.Contains(listed, want) {
problems = append(problems, fmt.Sprintf("%s's %s does not list the event %q in its `events` (design 47 G §529)", module, FindingEventsEntry, want))
}
}
for option, want := range map[string]string{"max-age-days": "7", "max-mib": "16"} {
if got := strings.TrimSpace(string(e.Options[option])); got != want {
problems = append(problems, fmt.Sprintf("%s's %s sets %s to %q, not %s (design 47 G §529)", module, FindingEventsEntry, option, got, want))
}
}
slices.Sort(problems)
return problems
}
// NodeEngineSeat is the seat the node-engine claims as one module (design 47 G §508).
const NodeEngineSeat = "node-engine"
// nodeEngineClaimProblems are the refusals of a revision's claims of the seat `node-engine`: one module claims it, a second
// is refused naming both (§508), and a revision in which none does is refused naming the seat (§527), since the
// node-engine's own module is read from the module whose entry claims it. A revision whose seat list lacks the seat is
// not judged.
func nodeEngineClaimProblems(shelf catalogue.Shelf) []string {
if _, known := catalogue.SeatNamed(NodeEngineSeat); !known {
return nil
}
var holders []string
for module, m := range shelf {
if slices.ContainsFunc(m.Claims, func(c catalogue.Claim) bool { return c.Name == NodeEngineSeat }) {
holders = append(holders, module)
}
}
slices.Sort(holders)
switch {
case len(holders) == 0:
return []string{fmt.Sprintf("no module of the revision claims the seat %s; the node-engine's own module is the one that does "+
"(design 47 G §527)", NodeEngineSeat)}
case len(holders) > 1:
return []string{fmt.Sprintf("%s claim the seat %s, which one module claims (design 47 G §508)", strings.Join(holders, " and "), NodeEngineSeat)}
}
return nil
}
@@ -170,6 +170,7 @@ func Between(shelf catalogue.Shelf, longestMachine int, known KnownPackages) []s
between = append(between, PackagesCatalogueProblems(shelf, known)...)
between = append(between, IdentityProblems(shelf, longestMachine)...)
between = append(between, DataProblems(shelf)...)
between = append(between, nodeEngineClaimProblems(shelf)...)
sort.Strings(between)
return slices.Compact(between)
}
@@ -2,13 +2,13 @@ package judge
import (
"encoding/json"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"reflect"
"strings"
"testing"
"git.novox.be/novox/mesh-delivery/sdk/composer"
"git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go/overlay"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
// The judge (novox/hq ADR 0578 §1): a manifest alone is accepted, or refused with each problem; an unknown field is
@@ -46,7 +46,7 @@ func TestTheJudge(t *testing.T) {
"enrolment": `{"module":"enrolment","version":"1.0","claims":[{"name":"enrolment","scope":"mesh","serves":["issue","again"]}],` +
`"tools":["enrolment_first_views","enrolment_first_view_again"]}`}},
} {
raws := map[string][]byte{}
raws := map[string][]byte{"node-engine": []byte(engineManifest("node-engine"))} // a revision has one claim of the seat node-engine (§527)
for k, v := range row.raws {
raws[k] = []byte(v)
}
@@ -153,3 +153,47 @@ func versioned(t *testing.T, raw []byte) []byte {
}
return out
}
// engineManifest is the manifest of a module that claims the seat node-engine, serving every verb the seat promises
// (design 47 G §508, §527).
func engineManifest(module string) string {
seat, _ := catalogue.SeatNamed(NodeEngineSeat)
var verbs []string
for _, v := range seat.Serves {
verbs = append(verbs, `"`+v.Name+`"`)
}
return `{"module":"` + module + `","version":"1.0","claims":[{"name":"node-engine","scope":"node","serves":[` + strings.Join(verbs, ",") + `]}]}`
}
// Design 47 G §508, §527: one module of a revision claims the seat node-engine; a second is refused naming both, and none
// refuses the revision naming the seat.
func TestTheRevisionHasOneClaimOfTheNodeEngine(t *testing.T) {
for _, row := range []struct {
name string
raws map[string]string
want string // empty: accepted
}{
{"one claim", map[string]string{"engine": engineManifest("engine")}, ""},
{"two claims", map[string]string{"engine": engineManifest("engine"), "other": engineManifest("other")},
"claim the seat node-engine"},
{"none", map[string]string{"ok": `{"module":"ok","version":"1.0"}`}, "claims the seat node-engine"},
} {
raws := map[string][]byte{}
for k, v := range row.raws {
raws[k] = []byte(v)
}
refused, unnamed := Shelf(raws, nil, 0)
got := strings.Join(append(unnamed, flat(refused)...), "\n")
if (row.want == "") != (got == "") || !strings.Contains(got, row.want) {
t.Errorf("%s: refused %q, want %q", row.name, got, row.want)
}
}
}
func flat(refused map[string][]string) []string {
var out []string
for _, p := range refused {
out = append(out, p...)
}
return out
}
@@ -16,13 +16,14 @@ import (
// OldSections are the sections of a manifest that keep bus state in the old forms.
var OldSections = []string{"state", "logs", "streams", "objects"}
// Foundation are the modules that keep using the old sections for good: exactly the ten parts the bus module's own
// Foundation are the modules that keep using the old sections for good: exactly the parts the bus module's own
// composition, assignments, build or delivery needs before it can answer (novox/hq ADR 0619 §1, §3, as ADR 0666 §4 changes
// them): the bus module, the grants part, the indexer, the assigner, the tool runner, the builder, the checker, mesh-config,
// the vault and the pipeline runner, which runs the bus module's own pipeline. Their new state goes on their old sections
// the vault, the pipeline runner, which runs the bus module's own pipeline, and the forge module and the image registry
// module, which its build calls (design 47 G §584). Their new state goes on their old sections
// too, so a foundation manifest asking bus state in the new form is refused.
var Foundation = []string{"nats", "mesh-grants", "mesh-delivery-indexer", "mesh-delivery-assigner", "node-tools",
"mesh-delivery-builder", "mesh-delivery-checker", "mesh-config", "mesh-vault", "mesh-delivery-pipeline"}
"mesh-delivery-builder", "mesh-delivery-checker", "mesh-config", "mesh-vault", "mesh-delivery-pipeline", "gitea", "distribution"}
// ViewOwners are the two owners whose state every reader reads through their first views and read tools, never by a
// `reads` entry (novox/hq ADR 0615 §1): CountOwnerReads counts what is left of those entries.
@@ -0,0 +1,43 @@
package judge
import (
"fmt"
"slices"
"strings"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// PackageNameProblems refuses a revision in which two modules' entries carry the same package name in one ecosystem, naming
// both (design 47 G §491, §676): a name read as a module must name one. It judges the entries' package fields, which the
// indexer wrote at take-in from the revision's own files, and answers each owning module's problems, sorted.
func PackageNameProblems(entries []meshcatalog.CatalogEntry) map[string][]string {
owners := map[string][]string{} // ecosystem and name -> modules
for _, e := range entries {
for _, a := range e.Artifacts {
if a.Package == nil {
continue
}
key := a.Package.Ecosystem + " " + a.Package.Name
if !slices.Contains(owners[key], e.Module) {
owners[key] = append(owners[key], e.Module)
}
}
}
out := map[string][]string{}
for key, modules := range owners {
if len(modules) < 2 {
continue
}
slices.Sort(modules)
ecosystem, name, _ := strings.Cut(key, " ")
problem := fmt.Sprintf("the %s package name %s is carried by the artifacts of %s: a package's name belongs to one module (design 47 G section 491)", ecosystem, name, strings.Join(modules, " and "))
for _, m := range modules {
out[m] = append(out[m], problem)
}
}
for m := range out {
slices.Sort(out[m])
}
return out
}
@@ -0,0 +1,41 @@
package judge
import (
"maps"
"reflect"
"slices"
"strings"
"testing"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// Design 47 G §491: two modules whose entries carry one Go module path are both refused, naming both; one name in two
// ecosystems, or one module's two artifacts of one name, is no problem.
func TestPackageNameProblems(t *testing.T) {
pkg := func(module, ecosystem, name string) meshcatalog.CatalogEntry {
return meshcatalog.CatalogEntry{Module: module, Artifacts: []meshcatalog.Artifact{{Ref: module + ":p", Package: &meshcatalog.ArtifactPackage{Ecosystem: ecosystem, Name: name}}}}
}
cases := []struct {
name string
entries []meshcatalog.CatalogEntry
want []string
}{
{"two modules, one Go name", []meshcatalog.CatalogEntry{pkg("a", "go", "x/y"), pkg("b", "go", "x/y"), pkg("c", "go", "x/z")}, []string{"a", "b"}},
{"one name, two ecosystems", []meshcatalog.CatalogEntry{pkg("a", "go", "x"), pkg("b", "npm", "x")}, nil},
{"one module twice", []meshcatalog.CatalogEntry{{Module: "a", Artifacts: []meshcatalog.Artifact{
{Ref: "a:p", Package: &meshcatalog.ArtifactPackage{Ecosystem: "go", Name: "x"}}, {Ref: "a:q", Package: &meshcatalog.ArtifactPackage{Ecosystem: "go", Name: "x"}}}}}, nil},
}
for _, c := range cases {
got := PackageNameProblems(c.entries)
modules := slices.Sorted(maps.Keys(got))
if !reflect.DeepEqual(modules, c.want) && (len(modules) > 0 || len(c.want) > 0) {
t.Errorf("%s: refused %v, want %v", c.name, modules, c.want)
}
for m, problems := range got {
if len(problems) != 1 || !strings.Contains(problems[0], strings.Join(c.want, " and ")) {
t.Errorf("%s: %s's problems %q do not name %v", c.name, m, problems, c.want)
}
}
}
}
@@ -4638,13 +4638,6 @@ func claimProblems(m catalogue.Manifest) []string {
}
for _, c := range m.Claims {
if s, known := catalogue.SeatNamed(c.Name); known && s.HeldByTheEngine {
// The node's engine holds it on every node, and no module may (novox/hq ADR 0563 §1): a module answering
// root-free or a hand-over would be an agent answering it.
problems = append(problems, fmt.Sprintf("%s claims %s, which every node's engine holds itself and no "+
"module may (novox/hq ADR 0563 §1)", m.Module, c.Name))
continue
}
if _, known := catalogue.SeatNamed(c.Name); known {
// **A seat's scope and what it delivers are not judged here** (novox/hq ADR 0122).
// This function runs wherever a manifest is parsed, and one of those places is the
+21 -14
View File
@@ -49,6 +49,7 @@ type stream struct {
Events []string `json:"events"`
MaxAgeDays int `json:"max-age-days"`
MaxMiB int `json:"max-mib"`
Retention string `json:"retention"`
}
type manifest struct {
@@ -122,7 +123,8 @@ 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.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17
{Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}, {Name: index.EntryIndexState, History: 1},
{Name: index.RegistrationsState, History: 1}, {Name: index.ProposalsState, 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,
@@ -155,10 +157,10 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
}
// "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control
// internal/broker/state.go:62-93).
// 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", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
// The registry seats' published fill each (§413: no consume of version.numbered) 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-conditions.findings-asked", "mesh-delivery-assigner.views.assignments",
"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", "npm-package-registry.published", "operator-channel.decided"}; !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" {
@@ -170,7 +172,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", "seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.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) {
if want := []string{"mesh-delivery-assigner.assignments", "seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "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
@@ -191,10 +193,17 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
index.SeatDefinitionTool, index.SeatDefinitionsAgainTool, index.OutOfStepTool, // DLV-I19
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) != 41 || slices.Contains(m.Tools, "modules") {
index.CatalogRevisionTool, index.CatalogRevisionsTool, index.CandidateGraphsTool, index.RecomputeTool, index.DropCandidateGraphTool,
index.EntryIndexTool, // design 47 G §675
index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool) // design 47 G §667, §8
if !reflect.DeepEqual(m.Tools, want) || len(want) != 45 || slices.Contains(m.Tools, "modules") {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
// Design 47 G §2, §9: every version of every registration is kept with no age limit, 64 MiB, on the old streams section.
if len(m.Streams) != 2 || m.Streams[0].Name != index.RegistrationHistoryStream || m.Streams[0].Retention != "history" || m.Streams[0].MaxMiB != 64 ||
m.Streams[1].Name != index.Module+"-findings" || m.Streams[1].MaxAgeDays != 7 || m.Streams[1].MaxMiB != 16 {
t.Errorf("streams = %+v, want registration-history keeping its history, 64 MiB, and the findings stream (§387)", m.Streams)
}
// Design 47 G §368: version is MAJOR.MINOR (0.1 while none is published, §368); §295, §299: the catalog revisions' form and
// the catch-up tool beside it (checked above).
if !regexp.MustCompile(`^[0-9]+\.[0-9]+$`).MatchString(m.Version) {
@@ -279,12 +288,10 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
}
}
}
// ADR 0635 §5: its views family is its ordinary events, `views.<state>` among its emits, and it has no stream of its own.
// ADR 0635 §5: its views family is its ordinary events, `views.<state>` among its emits, and its one stream is the
// registrations' history (design 47 G §2, asserted above).
// ADR 0636 §4: seat-definition-changed and -removed are the seat mesh-catalog's events, said on the seat's subject and kept
// on the shared events stream, and not among its own emits.
if len(m.Streams) != 0 {
t.Errorf("streams = %+v, want none", m.Streams)
}
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)
@@ -293,8 +300,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// ADR 0604 §2, ADR 0636 §4: it holds the seat mesh-catalog, serving its four verbs and seat-definitions from its tools
// bundle (which stays without a request subject); ADR 0663 §1: and revision, in the same runtime of its own.
if len(m.Claims) != 1 || m.Claims[0].Name != index.CatalogSeat || m.Claims[0].Scope != catalogue.ScopeMesh ||
!slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents", index.SeatDefinitionsVerb, "revision"}) {
t.Errorf("claims = %+v, want mesh-catalog with its six verbs", m.Claims)
!slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents", index.SeatDefinitionsVerb, "revision", "entry", "registrations", "register", "policy", "watch", "unwatch"}) {
t.Errorf("claims = %+v, want mesh-catalog with its seven verbs (entry: design 47 G §675)", m.Claims)
}
for _, e := range []string{index.SeatDefinitionChangedEvent, index.SeatDefinitionRemovedEvent} {
if slices.Contains(m.Emits, e) {
+23 -4
View File
@@ -13,7 +13,13 @@
"provides",
"dependents",
"seat-definitions",
"revision"
"revision",
"entry",
"registrations",
"register",
"policy",
"watch",
"unwatch"
]
}
],
@@ -58,20 +64,25 @@
"indexer_catalog_revisions",
"indexer_candidate_graphs",
"indexer_recompute",
"indexer_drop_candidate_graph"
"indexer_drop_candidate_graph",
"indexer_entries",
"indexer_proposals",
"indexer_proposal_drop",
"indexer_seed_again"
],
"consumes": [
"go-package-registry.published",
"mesh-build-go.toolchain.reported",
"mesh-build-node.toolchain.reported",
"mesh-conditions.findings-asked",
"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",
"operator-channel.decided",
"npm-package-registry.published"
],
"state": [
@@ -93,7 +104,14 @@
{ "name": "requirements", "history": 5 },
{ "name": "seat-definitions", "history": 5 },
{ "name": "pending", "history": 1 },
{ "name": "unjudged", "history": 1 }
{ "name": "unjudged", "history": 1 },
{ "name": "entry-index", "history": 1 },
{ "name": "registrations", "history": 1 },
{ "name": "proposals", "history": 1 }
],
"streams": [
{ "name": "registration-history", "events": ["registration.*"], "retention": "history", "max-mib": 64 },
{ "name": "mesh-delivery-indexer-findings", "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], "max-age-days": 7, "max-mib": 16 }
],
"objects": [
{ "name": "catalog-revisions", "max-mib": 512, "form": 1, "catch-up": "indexer_take_in" }
@@ -125,6 +143,7 @@
"seat:git.file",
"seat:git.contains",
"seat:git.open-pulls",
"seat:git.branch",
"mesh-delivery-checker.checker_tries",
"mesh-delivery-forge-bridge.bridge_merged_message",
"seat:node-engine.composition",