From 21c97051c8abb10920454dffc3280fc491749fb4 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 20:53:23 +0100 Subject: [PATCH] IX1-3: the registrations, the proposals that wait for their warrant, the seed, and the indexer's verbs and tools of them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The seat mesh-catalog's registrations, register, policy, watch and unwatch (each asks the operator, naming every value it moves, with the change's digest, and writes only on the warrant), the state proposals with an ask id and a digest per key, the consumer of operator-channel.decided, a restart that reads each open key, indexer_proposals, indexer_proposal_drop and indexer_seed_again, the seed as one ask. The router's record of an ask is one injected read that fails loudly until the design says how the program reads it (D-37); the seed's steps call is not wired. Builds: design 47 G §667, §2, §3, §4, §5, §8, §9, §10, §342, §343, §482 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-indexer-tools/main.go | 4 + .../cmd/mesh-delivery-indexer/main.go | 19 +- .../cmd/mesh-delivery-indexer/registry.go | 136 +++ .../cmd/mesh-delivery-indexer/takein.go | 5 +- modules/mesh-delivery-indexer/go.mod | 4 +- .../internal/index/candidates.go | 4 + .../internal/index/jcs.go | 133 +++ .../internal/index/pending.go | 4 +- .../internal/index/pending_test.go | 2 +- .../internal/index/registered.go | 3 + .../internal/index/registrations.go | 921 ++++++++++++++++++ .../internal/index/registrations_test.go | 294 ++++++ .../internal/index/registrations_tools.go | 119 +++ .../internal/index/seed.go | 171 ++++ .../internal/index/takein.go | 19 +- .../mesh-delivery-indexer/manifest_test.go | 25 +- modules/mesh-delivery-indexer/module.json | 21 +- 17 files changed, 1858 insertions(+), 26 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go create mode 100644 modules/mesh-delivery-indexer/internal/index/jcs.go create mode 100644 modules/mesh-delivery-indexer/internal/index/registrations.go create mode 100644 modules/mesh-delivery-indexer/internal/index/registrations_test.go create mode 100644 modules/mesh-delivery-indexer/internal/index/registrations_tools.go create mode 100644 modules/mesh-delivery-indexer/internal/index/seed.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index 5de2e524..23c30b89 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -293,6 +293,10 @@ func main() { _, 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 { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index f2e5722a..cb57a985 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -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. @@ -634,12 +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 } } - _ = 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 { + // 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 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 diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go new file mode 100644 index 00000000..176b0c6d --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go @@ -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() +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index 5cabeaa1..a1c9d068 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -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 { diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 86f555fb..d636d4fe 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -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 diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 2ba46b92..cfb56164 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -107,6 +107,10 @@ type Indexer struct { // 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 diff --git a/modules/mesh-delivery-indexer/internal/index/jcs.go b/modules/mesh-delivery-indexer/internal/index/jcs.go new file mode 100644 index 00000000..701893e1 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/jcs.go @@ -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('"') +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 2f93aa52..36a8b923 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -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, EntryIndex: 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) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index b8f929e9..b5d7747c 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -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", "EntryIndex"} { + "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/registered.go b/modules/mesh-delivery-indexer/internal/index/registered.go index 5bf426d7..ce5fc2aa 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered.go +++ b/modules/mesh-delivery-indexer/internal/index/registered.go @@ -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 diff --git a/modules/mesh-delivery-indexer/internal/index/registrations.go b/modules/mesh-delivery-indexer/internal/index/registrations.go new file mode 100644 index 00000000..f4f4dc40 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/registrations.go @@ -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: ., and for watch and unwatch .., 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 .; 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, ®); 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, ®) == 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 . 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 .), 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) +} diff --git a/modules/mesh-delivery-indexer/internal/index/registrations_test.go b/modules/mesh-delivery-indexer/internal/index/registrations_test.go new file mode 100644 index 00000000..4b0b91a0 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/registrations_test.go @@ -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 := ®Rig{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") + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/registrations_tools.go b/modules/mesh-delivery-indexer/internal/index/registrations_tools.go new file mode 100644 index 00000000..290b6847 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/registrations_tools.go @@ -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 (., 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 +} diff --git a/modules/mesh-delivery-indexer/internal/index/seed.go b/modules/mesh-delivery-indexer/internal/index/seed.go new file mode 100644 index 00000000..d25a4218 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/seed.go @@ -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)) +} diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index f325dccf..89ca3313 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -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 diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 439123d8..1bb86b93 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -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}, {Name: index.EntryIndexState, 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, @@ -158,7 +160,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // 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-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"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + "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 @@ -192,10 +194,15 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build 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, - index.EntryIndexTool) // design 47 G §675 - if !reflect.DeepEqual(m.Tools, want) || len(want) != 42 || slices.Contains(m.Tools, "modules") { + 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) != 1 || m.Streams[0].Name != index.RegistrationHistoryStream || m.Streams[0].Retention != "history" || m.Streams[0].MaxMiB != 64 { + t.Errorf("streams = %+v, want registration-history keeping its history, 64 MiB", 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) { @@ -280,12 +287,10 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } } } - // ADR 0635 §5: its views family is its ordinary events, `views.` among its emits, and it has no stream of its own. + // ADR 0635 §5: its views family is its ordinary events, `views.` 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) @@ -294,7 +299,7 @@ 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", "entry"}) { + !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} { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 5c4096db..ffe7fcb8 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -14,7 +14,12 @@ "dependents", "seat-definitions", "revision", - "entry" + "entry", + "registrations", + "register", + "policy", + "watch", + "unwatch" ] } ], @@ -60,7 +65,10 @@ "indexer_candidate_graphs", "indexer_recompute", "indexer_drop_candidate_graph", - "indexer_entries" + "indexer_entries", + "indexer_proposals", + "indexer_proposal_drop", + "indexer_seed_again" ], "consumes": [ "go-package-registry.published", @@ -73,6 +81,7 @@ "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", "mesh-delivery.source-moved", + "operator-channel.decided", "npm-package-registry.published" ], "state": [ @@ -95,7 +104,12 @@ { "name": "seat-definitions", "history": 5 }, { "name": "pending", "history": 1 }, { "name": "unjudged", "history": 1 }, - { "name": "entry-index", "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 } ], "objects": [ { "name": "catalog-revisions", "max-mib": 512, "form": 1, "catch-up": "indexer_take_in" } @@ -127,6 +141,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",