IX1-3: the registrations, the proposals that wait for their warrant, the seed, and the indexer's verbs and tools of them

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
This commit is contained in:
2026-11-11 20:53:23 +01:00
parent 1aeccdfb01
commit 21c97051c8
17 changed files with 1858 additions and 26 deletions
@@ -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 {
@@ -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
@@ -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()
}
@@ -27,6 +27,7 @@ type takerIn interface {
FingerprintsAgain(ctx context.Context, module string) index.TakenIn
RecomputeByHand(ctx context.Context, by, why string) index.TakenIn
DropCandidateGraph(ctx context.Context, group, heads, by, why string) index.TakenIn
RegistryAct(ctx context.Context, a index.TakeInAsk) index.TakenIn
}
// takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the
@@ -45,7 +46,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
return
}
asked, r := "take-in", index.TakenIn{}
if a.Recompute {
if a.Registry != nil {
asked, r = "registrations' act "+a.Registry.Verb, ix.RegistryAct(ctx, a)
} else if a.Recompute {
asked, r = "recompute of the catalog graph", ix.RecomputeByHand(ctx, a.By, a.Why)
a.Module = r.Module
} else if a.CandidateGroup != "" && a.Drop {
+3 -1
View File
@@ -9,8 +9,8 @@ require (
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/git v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker v0.0.0-00010101000000-000000000000
@@ -70,6 +70,8 @@ require (
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry v0.0.0-00010101000000-000000000000 // indirect
@@ -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
@@ -0,0 +1,133 @@
package index
import (
"bytes"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"sort"
"strings"
"unicode/utf16"
)
// Canonical is a JSON value in RFC 8785's canonical form: keys sorted by their UTF-16 code units, no white space, strings
// escaped as the RFC says, integers as digits. A number with a fraction or an exponent is refused: no change the indexer
// digests holds one.
func Canonical(v any) ([]byte, error) {
raw, err := json.Marshal(v)
if err != nil {
return nil, err
}
dec := json.NewDecoder(bytes.NewReader(raw))
dec.UseNumber()
var tree any
if err := dec.Decode(&tree); err != nil {
return nil, err
}
var b bytes.Buffer
if err := canonical(&b, tree); err != nil {
return nil, err
}
return b.Bytes(), nil
}
// DigestOf is "sha256:" and the digits of the canonical form of v.
func DigestOf(v any) (string, error) {
c, err := Canonical(v)
if err != nil {
return "", err
}
sum := sha256.Sum256(c)
return "sha256:" + hex.EncodeToString(sum[:]), nil
}
func canonical(b *bytes.Buffer, v any) error {
switch t := v.(type) {
case nil:
b.WriteString("null")
case bool:
if t {
b.WriteString("true")
} else {
b.WriteString("false")
}
case json.Number:
if strings.ContainsAny(string(t), ".eE") {
return fmt.Errorf("the number %s has a fraction or an exponent, which the canonical form of a change does not hold", t)
}
b.WriteString(string(t))
case string:
canonicalString(b, t)
case []any:
b.WriteByte('[')
for i, e := range t {
if i > 0 {
b.WriteByte(',')
}
if err := canonical(b, e); err != nil {
return err
}
}
b.WriteByte(']')
case map[string]any:
keys := make([]string, 0, len(t))
for k := range t {
keys = append(keys, k)
}
sort.Slice(keys, func(i, j int) bool { return slices16(keys[i], keys[j]) })
b.WriteByte('{')
for i, k := range keys {
if i > 0 {
b.WriteByte(',')
}
canonicalString(b, k)
b.WriteByte(':')
if err := canonical(b, t[k]); err != nil {
return err
}
}
b.WriteByte('}')
default:
return fmt.Errorf("a value of type %T has no canonical form", v)
}
return nil
}
// slices16 orders two strings by their UTF-16 code units, as RFC 8785 sorts keys.
func slices16(a, b string) bool {
x, y := utf16.Encode([]rune(a)), utf16.Encode([]rune(b))
for i := 0; i < len(x) && i < len(y); i++ {
if x[i] != y[i] {
return x[i] < y[i]
}
}
return len(x) < len(y)
}
func canonicalString(b *bytes.Buffer, s string) {
b.WriteByte('"')
for _, r := range s {
switch {
case r == '"':
b.WriteString(`\"`)
case r == '\\':
b.WriteString(`\\`)
case r == '\b':
b.WriteString(`\b`)
case r == '\f':
b.WriteString(`\f`)
case r == '\n':
b.WriteString(`\n`)
case r == '\r':
b.WriteString(`\r`)
case r == '\t':
b.WriteString(`\t`)
case r < 0x20:
fmt.Fprintf(b, `\u%04x`, r)
default:
b.WriteRune(r)
}
}
b.WriteByte('"')
}
@@ -910,7 +910,9 @@ func (ix *Indexer) dry() *dryRun {
Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil,
OwnBuildState: nil, underSeats: true,
// A dry run keeps no catalog graph revision (catalog_revisions.go): its revision is the take-in's to keep.
Revisions: nil, SayCatalogRevision: nil, SayFill: nil, 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)
@@ -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{})
@@ -2,6 +2,7 @@ package index
import (
"context"
"encoding/json"
"fmt"
"maps"
"slices"
@@ -224,6 +225,8 @@ type TakenIn struct {
Unread []string `json:"unread,omitempty"`
// Fingerprints are the modules whose fingerprint a recompute by hand moved (FingerprintsAgain), each old and new.
Fingerprints []FingerprintMove `json:"fingerprints,omitempty"`
// Registry is the answer of a registrations' act (RegistryAct): the verb's own answer, as JSON.
Registry json.RawMessage `json:"registry,omitempty"`
}
// TakeIn reads a registered module's manifest now, whatever was tried before, and writes it as a merge would (the tool
@@ -0,0 +1,921 @@
package index
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"sort"
"strings"
"sync"
"time"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
gitseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/git"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// A module's registration, and the trusted change of one (design 47 G §2 to §4, §10, §342, §343, §667): the indexer keeps each
// module's newest registration in its old state section registrations, one key per module, every version on its old
// streams section's stream registration-history, and writes a version only on the operator's warrant. A change waits for
// its warrant in the old state section proposals.
const (
// RegistrationsState is the old state section of the registrations, one key per module (§2).
RegistrationsState = "registrations"
// RegistrationsBucket is the bucket of the state.
RegistrationsBucket = Module + "_" + RegistrationsState
// ProposalsState is the old state section of the changes that wait for their warrant (§667), capped at 4 MiB.
ProposalsState = "proposals"
// ProposalsBucket is the bucket of the state.
ProposalsBucket = Module + "_" + ProposalsState
// RegistrationHistoryStream is the old streams section that keeps every version of every registration (§2, §9).
RegistrationHistoryStream = "registration-history"
// RegistrationHistoryEvent is the event of the stream, one subject per module.
RegistrationHistoryEvent = "registration"
// SeedKey is the key of the first registrations' proposal (§8).
SeedKey = "seed"
// ProposalsTool, ProposalDropTool and SeedAgainTool are the tools of the proposals (§667, §8).
ProposalsTool = "indexer_proposals"
ProposalDropTool = "indexer_proposal_drop"
SeedAgainTool = "indexer_seed_again"
// ProposalLasts is how long the operator has to answer: the router's bound for an approval less ten minutes.
ProposalLasts = occ.ApproveLasts - 10*time.Minute
)
// The states of a proposal's key (§667): open while its warrant is awaited, then how it ended.
const (
ProposalOpen = "open"
ProposalApplied = "applied"
ProposalDeclined = "declined"
ProposalExpired = "expired"
ProposalRefused = "refused"
ProposalDropped = "dropped"
ProposalLost = "lost"
// ProposalSeedWaits is the seed's end that keeps the finding seed-waits (§8).
ProposalSeedWaits = "seed-waits"
)
const (
optionApprove = "approve"
optionDecline = "decline"
)
// The verbs of a trusted change.
const (
verbRegister = "register"
verbPolicy = "policy"
verbWatch = "watch"
verbUnwatch = "unwatch"
verbSeed = "seed"
)
// Proposal is one value of the state proposals: a change awaiting its warrant, whole, with its digest, the current ask's id,
// its caller and when it was asked; ended, it keeps its outcome (§667).
type Proposal struct {
V int `json:"v"`
Verb string `json:"verb"`
Module string `json:"module,omitempty"`
Branch string `json:"branch,omitempty"`
Change json.RawMessage `json:"change"`
// Was is the version of the registration the change was composed against (0 for none): a change warranted when the
// registration moved is refused, never written over the newer one.
Was int64 `json:"was"`
Lines []string `json:"lines"`
Digest string `json:"digest"`
Ask string `json:"ask"`
Caller string `json:"caller,omitempty"`
Why string `json:"why,omitempty"`
At time.Time `json:"at"`
// State is open, or how the key ended; Outcome says it in words; Ended is when.
State string `json:"state"`
Outcome string `json:"outcome,omitempty"`
Ended time.Time `json:"ended,omitzero"`
// Seed holds, for the seed's key, the registrations it proposes (§8).
Seed []meshcatalog.Registration `json:"seed,omitempty"`
}
// ProposalKey is the key of a change: <verb>.<module>, and for watch and unwatch <verb>.<module>.<branch>, each token
// escaped as a key's name is (§667).
func ProposalKey(verb, module, branch string) string {
k := verb + "." + keys.EscapeName(module)
if branch != "" {
k += "." + keys.EscapeName(branch)
}
return k
}
// RegistrationKey is a module's key in registrations.
func RegistrationKey(module string) string { return keys.EscapeName(module) }
// Registry is the registrations and the trusted changes of them. Its parts are the indexer's: the two states, the history
// stream, the event, the seat git's branch, the seat operator-channel's ask.
type Registry struct {
Registrations Store
Proposals Store
// History keeps one version on the stream registration-history, once under its key <module>.<version>; HistoryOf reads
// every version of one module, oldest first.
History func(ctx context.Context, r meshcatalog.Registration) error
HistoryOf func(ctx context.Context, module string) ([]meshcatalog.Registration, error)
// Say says the seat mesh-catalog's event registration-changed.
Say func(ctx context.Context, e meshcatalog.RegistrationChangedEvent) error
// Branch reads a branch through the seat git; an empty branch asks the repository's default branch.
Branch func(repository, branch string) (gitseat.BranchAnswer, error)
// Ask publishes an ask under the indexer's own name, Cancel takes it back, Where reads the router's own record of an
// ask, read under the indexer's module's name (state, and the warrant when it holds one; none when it holds no ask).
Ask func(a oc.Ask) error
Cancel func(id string) error
Where func(id string) (state string, w *oc.Warrant, err error)
// SeedWaits says the finding seed-waits naming the ask, and clears it (§8); nil says nothing.
SeedWaits func(ask string, waiting bool)
// Seed composes the first registrations (§8); nil proposes none.
Seed func(ctx context.Context) (SeedProposal, error)
Now func() time.Time
Logf func(string, ...any)
mu sync.Mutex
}
func (r *Registry) now() time.Time {
if r.Now != nil {
return r.Now().UTC()
}
return time.Now().UTC()
}
func (r *Registry) logf(format string, args ...any) {
if r.Logf != nil {
r.Logf(format, args...)
}
}
// Registration is one module's newest registration.
func (r *Registry) Registration(ctx context.Context, module string) (meshcatalog.Registration, bool, error) {
v, _, found, err := r.Registrations.Get(ctx, RegistrationKey(module))
if err != nil || !found {
return meshcatalog.Registration{}, false, err
}
var reg meshcatalog.Registration
if err := json.Unmarshal(v, &reg); err != nil {
return reg, false, fmt.Errorf("the registration of %s cannot be read: %w", module, err)
}
return reg, true, nil
}
// Registrations is the verb registrations (§4): every registration, or one module's, with its history.
func (r *Registry) RegistrationsAnswer(ctx context.Context, in meshcatalog.RegistrationsInput) (meshcatalog.RegistrationsAnswer, error) {
out := meshcatalog.RegistrationsAnswer{Registrations: []meshcatalog.Registration{}}
if in.Module != "" {
reg, found, err := r.Registration(ctx, in.Module)
if err != nil {
return out, err
}
if found {
out.Registrations = append(out.Registrations, reg)
}
if in.History && r.HistoryOf != nil {
if out.History, err = r.HistoryOf(ctx, in.Module); err != nil {
return out, fmt.Errorf("the history of %s cannot be read: %w", in.Module, err)
}
}
return out, nil
}
all, err := r.Registrations.Keys(ctx)
if err != nil {
return out, err
}
slices.Sort(all)
for _, k := range all {
v, _, found, err := r.Registrations.Get(ctx, k)
if err != nil {
return out, err
}
var reg meshcatalog.Registration
if found && json.Unmarshal(v, &reg) == nil {
out.Registrations = append(out.Registrations, reg)
}
}
return out, nil
}
// refused is the answer of a change that raises no ask.
func refused(module, reason, words string) meshcatalog.RegistrationAsked {
return meshcatalog.RegistrationAsked{Module: module, Refused: reason, Words: words}
}
// forbiddenBranch refuses the two names the catalog's keys use (§343).
func forbiddenBranch(b string) string {
if b == "candidate" || b == "entry" {
return fmt.Sprintf("the branch %q cannot be watched: the catalog keeps its revisions under <branch>.<revision> beside the key forms candidate and entry", b)
}
return ""
}
// reviewRequired refuses code review as a required step (§3, §482).
func reviewRequired(s meshcatalog.StepStatuses) bool {
return s.Review == meshcatalog.StepStatusesApprovalRequired
}
// protect refuses a watched branch the forge does not protect (§5), naming it.
func (r *Registry) protect(repository string, branches []string) (string, error) {
for _, b := range branches {
if w := forbiddenBranch(b); w != "" {
return w, nil
}
a, err := r.Branch(repository, b)
if err != nil {
return "", fmt.Errorf("the seat git does not answer for %s %s: %w", repository, b, err)
}
if a.Error != "" {
return "", fmt.Errorf("the seat git cannot read %s %s: %s", repository, b, a.Error)
}
if !a.Protected {
return fmt.Sprintf("the branch %s of %s is not protected: no rule refuses a direct push and admits only merges", b, repository), nil
}
}
return "", nil
}
// stepLines names each step status a policy moves, one line each.
func stepLines(prefix string, old, next meshcatalog.StepStatuses) []string {
var lines []string
for _, s := range []struct {
name string
was, now string
}{{"approval", old.Approval, next.Approval}, {"composition", old.Composition, next.Composition},
{"integration", old.Integration, next.Integration}, {"lint", old.Lint, next.Lint},
{"code review", old.Review, next.Review}, {"unit tests", old.Unit, next.Unit}} {
if s.was != s.now {
lines = append(lines, fmt.Sprintf("%s%s: %s -> %s", prefix, s.name, cmpEmpty(s.was), cmpEmpty(s.now)))
}
}
return lines
}
func cmpEmpty(s string) string {
if s == "" {
return "none"
}
return s
}
// overlay puts the statuses a policy names over the ones the module keeps.
func overlay(base, over meshcatalog.StepStatuses) meshcatalog.StepStatuses {
for _, p := range []struct {
b *string
o string
}{{&base.Approval, over.Approval}, {&base.Composition, over.Composition},
{&base.Integration, over.Integration}, {&base.Lint, over.Lint}, {&base.Review, over.Review}, {&base.Unit, over.Unit}} {
if p.o != "" {
*p.b = p.o
}
}
return base
}
// compose is the registration a change makes of the one kept (none: found false), and the lines that say each value it moves.
func compose(verb string, cur meshcatalog.Registration, found bool, change json.RawMessage) (meshcatalog.Registration, []string, error) {
switch verb {
case verbRegister:
var in meshcatalog.RegisterInput
if err := json.Unmarshal(change, &in); err != nil {
return cur, nil, err
}
next := in.Registration
next.Version, next.Warrant, next.By, next.At = 0, "", "", nil
var lines []string
if !found {
lines = append(lines, fmt.Sprintf("%s: registered from %s at %s, watching %s", next.Module, next.Origin, next.Path, strings.Join(next.Branches, ", ")))
lines = append(lines, stepLines(next.Module+" ", meshcatalog.StepStatuses{}, next.Policy.Steps)...)
return next, lines, nil
}
if cur.Origin != next.Origin {
lines = append(lines, fmt.Sprintf("%s: origin %s -> %s", next.Module, cur.Origin, next.Origin))
}
if cur.Path != next.Path {
lines = append(lines, fmt.Sprintf("%s: manifest path %s -> %s", next.Module, cur.Path, next.Path))
}
for _, b := range next.Branches {
if !slices.Contains(cur.Branches, b) {
lines = append(lines, fmt.Sprintf("%s: watches %s", next.Module, b))
}
}
for _, b := range cur.Branches {
if !slices.Contains(next.Branches, b) {
lines = append(lines, fmt.Sprintf("%s: stops watching %s", next.Module, b))
}
}
lines = append(lines, stepLines(next.Module+" ", cur.Policy.Steps, next.Policy.Steps)...)
lines = append(lines, artifactLines(next.Module, cur.Policy, next.Policy)...)
return next, lines, nil
case verbPolicy:
var in meshcatalog.PolicyInput
if err := json.Unmarshal(change, &in); err != nil {
return cur, nil, err
}
next := cur
var lines []string
if in.Artifact != "" {
next.Policy.Artifacts = map[string]meshcatalog.StepStatuses{}
for k, v := range cur.Policy.Artifacts {
next.Policy.Artifacts[k] = v
}
was := cur.Policy.Artifacts[in.Artifact]
now := overlay(was, in.Steps)
next.Policy.Artifacts[in.Artifact] = now
lines = stepLines(fmt.Sprintf("%s:%s ", in.Module, in.Artifact), was, now)
} else {
next.Policy.Steps = overlay(cur.Policy.Steps, in.Steps)
lines = stepLines(in.Module+" ", cur.Policy.Steps, next.Policy.Steps)
}
return next, lines, nil
case verbWatch, verbUnwatch:
var in meshcatalog.WatchInput
if err := json.Unmarshal(change, &in); err != nil {
return cur, nil, err
}
next := cur
next.Branches = slices.Clone(cur.Branches)
var lines []string
if verb == verbWatch && !slices.Contains(cur.Branches, in.Branch) {
next.Branches = append(next.Branches, in.Branch)
lines = append(lines, fmt.Sprintf("%s: watches %s", in.Module, in.Branch))
}
if verb == verbUnwatch && slices.Contains(cur.Branches, in.Branch) {
next.Branches = slices.DeleteFunc(next.Branches, func(b string) bool { return b == in.Branch })
lines = append(lines, fmt.Sprintf("%s: stops watching %s", in.Module, in.Branch))
}
return next, lines, nil
}
return cur, nil, fmt.Errorf("%q is no change of a registration", verb)
}
func artifactLines(module string, old, next meshcatalog.Policy) []string {
names := map[string]bool{}
for k := range old.Artifacts {
names[k] = true
}
for k := range next.Artifacts {
names[k] = true
}
sorted := make([]string, 0, len(names))
for k := range names {
sorted = append(sorted, k)
}
sort.Strings(sorted)
var lines []string
for _, a := range sorted {
lines = append(lines, stepLines(module+":"+a+" ", old.Artifacts[a], next.Artifacts[a])...)
}
return lines
}
// Propose is the common path of the four verbs (§3): it checks the change, composes the lines it moves, keeps it in proposals
// under its key (a newer change of the same key replaces the open ask, ADR 0666 §12) and asks the operator, naming every value
// it moves and carrying the change's digest. It writes no registration.
func (r *Registry) Propose(ctx context.Context, verb, caller string, change json.RawMessage) (meshcatalog.RegistrationAsked, error) {
var module, branch, why string
switch verb {
case verbRegister:
var in meshcatalog.RegisterInput
if err := json.Unmarshal(change, &in); err != nil {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("register takes a registration: %w", err)
}
module, why = in.Registration.Module, in.Why
case verbPolicy:
var in meshcatalog.PolicyInput
if err := json.Unmarshal(change, &in); err != nil {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("policy takes a module and its steps: %w", err)
}
module, why = in.Module, in.Why
case verbWatch, verbUnwatch:
var in meshcatalog.WatchInput
if err := json.Unmarshal(change, &in); err != nil {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("%s takes a module and a branch: %w", verb, err)
}
module, branch, why = in.Module, in.Branch, in.Why
default:
return meshcatalog.RegistrationAsked{}, fmt.Errorf("%q is no change of a registration", verb)
}
if strings.TrimSpace(module) == "" {
return meshcatalog.RegistrationAsked{}, errors.New("name the module (module)")
}
if (verb == verbWatch || verb == verbUnwatch) && branch == "" {
return meshcatalog.RegistrationAsked{}, errors.New("name the branch (branch)")
}
r.mu.Lock()
defer r.mu.Unlock()
cur, found, err := r.Registration(ctx, module)
if err != nil {
return meshcatalog.RegistrationAsked{}, err
}
if verb != verbRegister && !found {
return refused(module, meshcatalog.RegistrationAskedRefusedUnknownModule, module+" is not registered: register it first"), nil
}
// register with no branch takes the repository's default branch (§342).
if verb == verbRegister {
var in meshcatalog.RegisterInput
_ = json.Unmarshal(change, &in)
if in.Registration.Origin == "" || in.Registration.Path == "" {
return meshcatalog.RegistrationAsked{}, errors.New("name the registration's origin and the path of its manifest")
}
if len(in.Registration.Branches) == 0 {
a, err := r.Branch(in.Registration.Origin, "")
if err != nil {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("the seat git does not answer the default branch of %s: %w", in.Registration.Origin, err)
}
if a.Error != "" || a.Branch == "" {
return meshcatalog.RegistrationAsked{}, fmt.Errorf("the default branch of %s is not read: %s", in.Registration.Origin, a.Error)
}
in.Registration.Branches = []string{a.Branch}
if change, err = json.Marshal(in); err != nil {
return meshcatalog.RegistrationAsked{}, err
}
}
}
steps := []meshcatalog.StepStatuses{}
switch verb {
case verbRegister:
var in meshcatalog.RegisterInput
_ = json.Unmarshal(change, &in)
steps = append(steps, in.Registration.Policy.Steps)
for _, a := range in.Registration.Policy.Artifacts {
steps = append(steps, a)
}
case verbPolicy:
var in meshcatalog.PolicyInput
_ = json.Unmarshal(change, &in)
steps = append(steps, in.Steps)
}
for _, s := range steps {
if reviewRequired(s) {
return refused(module, meshcatalog.RegistrationAskedRefusedReviewRequired,
"code review is never a required step: no record binds a forge account to its bus identity, so no review can gate while its author cannot be told"), nil
}
}
next, lines, err := compose(verb, cur, found, change)
if err != nil {
return meshcatalog.RegistrationAsked{}, err
}
var branches []string
switch verb {
case verbRegister:
branches = next.Branches
case verbWatch:
branches = []string{branch}
}
if len(branches) > 0 {
words, err := r.protect(next.Origin, branches)
if err != nil {
return meshcatalog.RegistrationAsked{}, err
}
if words != "" {
return refused(module, meshcatalog.RegistrationAskedRefusedUnprotectedBranch, words), nil
}
}
if len(lines) == 0 {
return refused(module, meshcatalog.RegistrationAskedRefusedNothingMoves, "the change moves no value of "+module+"'s registration"), nil
}
p := Proposal{V: 1, Verb: verb, Module: module, Branch: branch, Change: change, Lines: lines, Caller: caller, Why: why, At: r.now(), State: ProposalOpen}
if found {
p.Was = cur.Version
}
if p.Digest, err = DigestOf(map[string]any{"verb": verb, "module": module, "change": json.RawMessage(change)}); err != nil {
return meshcatalog.RegistrationAsked{}, err
}
if err := r.keepAndAsk(ctx, ProposalKey(verb, module, branch), &p); err != nil {
return meshcatalog.RegistrationAsked{}, err
}
return meshcatalog.RegistrationAsked{Module: module, Ask: p.Ask, Lines: lines}, nil
}
// askID is the id of the ask of a change: it names the key's digest and time, so a newer change of the key is a new id.
func askID(key, digest string, at time.Time) string {
d := strings.TrimPrefix(digest, "sha256:")
if len(d) > 16 {
d = d[:16]
}
return fmt.Sprintf("reg-%s-%d", d, at.Unix())
}
// composeAsk is the ask of a proposal, composed again from it wherever it is checked.
func (p Proposal) composeAsk(key string) oc.Ask {
headline := "Change the registration of " + p.Module + "?"
if p.Verb == verbSeed {
headline = "Register the first modules?"
}
if r := []rune(headline); len(r) > occ.HeadlineLength {
headline = string(r[:occ.HeadlineLength-1]) + "…"
}
explain := strings.Join(p.Lines, "\n")
if p.Why != "" {
explain += "\nWhy: " + p.Why
}
explain += "\nDigest " + p.Digest
a := oc.Ask{ID: p.Ask, Headline: headline, Explanation: explain, Who: occ.Operator,
Expires: occ.FormatTime(p.At.Add(ProposalLasts)), OnExpiry: "nothing is written; the change waits to be proposed again",
About: "catalog.proposal." + key}
for _, o := range []struct{ id, label, does, level string }{
{optionApprove, "Approve", "the registration is written as these lines say", occ.Approve},
{optionDecline, "Decline", "nothing is written", occ.Acknowledge},
} {
opt := oc.Option{ID: o.id, Label: o.label, Does: o.does, Level: o.level}
if o.level != occ.Acknowledge {
opt.Binds, _ = occ.ActDigest(p.act())
}
a.Options = append(a.Options, opt)
}
return a
}
// act is what the Approve binds: the change's verb, its module and its digest (ADR 0286 §10).
func (p Proposal) act() occ.Act {
return occ.Act{"verb": p.Verb, "module": p.Module, "digest": p.Digest}
}
// keepAndAsk keeps the proposal under its key and publishes its ask; a newer change of the key replaces the open ask.
func (r *Registry) keepAndAsk(ctx context.Context, key string, p *Proposal) error {
p.Ask = askID(key, p.Digest, p.At)
ask := p.composeAsk(key)
if err := occ.Check(ask, p.At); err != nil {
return err
}
data, err := json.Marshal(p)
if err != nil {
return err
}
if err := write(ctx, r.Proposals, key, func(old []byte, found bool) ([]byte, bool, error) {
if found {
var o Proposal
if json.Unmarshal(old, &o) == nil && o.State == ProposalOpen && o.Ask == p.Ask {
return nil, false, fmt.Errorf("the same change is asked already, as %s", o.Ask)
}
}
return data, false, nil
}); err != nil {
return fmt.Errorf("the change could not be kept, so the operator was not asked: %w", err)
}
if err := r.Ask(ask); err != nil {
p.State, p.Outcome, p.Ended = ProposalRefused, "the ask could not be published: "+err.Error(), r.now()
_ = r.keep(ctx, key, p)
return fmt.Errorf("the operator could not be asked: %w", err)
}
return nil
}
func (r *Registry) keep(ctx context.Context, key string, p *Proposal) error {
data, err := json.Marshal(p)
if err != nil {
return err
}
return write(ctx, r.Proposals, key, func([]byte, bool) ([]byte, bool, error) { return data, false, nil })
}
// ProposalRow is one key of proposals as indexer_proposals lists it.
type ProposalRow struct {
Key string `json:"key"`
Verb string `json:"verb"`
Module string `json:"module,omitempty"`
State string `json:"state"`
Ask string `json:"ask"`
Caller string `json:"caller,omitempty"`
At time.Time `json:"at"`
Outcome string `json:"outcome,omitempty"`
Ended time.Time `json:"ended,omitzero"`
Lines []string `json:"lines,omitempty"`
Digest string `json:"digest"`
}
// ListProposals is the open and the ended keys with their asks and outcomes (indexer_proposals, class read).
func ListProposals(ctx context.Context, st Store) ([]ProposalRow, error) {
all, err := st.Keys(ctx)
if err != nil {
return nil, err
}
slices.Sort(all)
out := []ProposalRow{}
for _, k := range all {
v, _, found, err := st.Get(ctx, k)
if err != nil {
return nil, err
}
var p Proposal
if !found || json.Unmarshal(v, &p) != nil {
continue
}
out = append(out, ProposalRow{Key: k, Verb: p.Verb, Module: p.Module, State: p.State, Ask: p.Ask, Caller: p.Caller, At: p.At,
Outcome: p.Outcome, Ended: p.Ended, Lines: p.Lines, Digest: p.Digest})
}
return out, nil
}
func (r *Registry) load(ctx context.Context, key string) (Proposal, uint64, bool, error) {
v, rev, found, err := r.Proposals.Get(ctx, key)
if err != nil || !found {
return Proposal{}, 0, found, err
}
var p Proposal
if err := json.Unmarshal(v, &p); err != nil {
return p, rev, true, fmt.Errorf("the proposal %s cannot be read: %w", key, err)
}
return p, rev, true, nil
}
// end writes how a key ended, when it is still the open one of this ask.
func (r *Registry) end(ctx context.Context, key, ask, state, outcome string) error {
return write(ctx, r.Proposals, key, func(old []byte, found bool) ([]byte, bool, error) {
var p Proposal
if !found || json.Unmarshal(old, &p) != nil || p.Ask != ask || p.State != ProposalOpen {
return nil, false, nil
}
p.State, p.Outcome, p.Ended = state, outcome, r.now()
data, err := json.Marshal(p)
return data, false, err
})
}
// Drop is indexer_proposal_drop (class act): it cancels the key's open ask and ends the key dropped.
func (r *Registry) Drop(ctx context.Context, key, why, by string) (string, error) {
if strings.TrimSpace(why) == "" {
return "", errors.New("say why (why): a hand act is said in the hand-act log")
}
r.mu.Lock()
defer r.mu.Unlock()
p, _, found, err := r.load(ctx, key)
switch {
case err != nil:
return "", err
case !found:
return "", fmt.Errorf("no proposal %q is kept; indexer_proposals lists the keys", key)
case p.State != ProposalOpen:
return "", fmt.Errorf("the proposal %q ended %s, and has no open ask to drop", key, p.State)
}
if err := r.Cancel(p.Ask); err != nil {
return "", fmt.Errorf("the ask %s could not be cancelled: %w", p.Ask, err)
}
if err := r.end(ctx, key, p.Ask, ProposalDropped, fmt.Sprintf("dropped by %s: %s", by, why)); err != nil {
return "", err
}
if key == SeedKey && r.SeedWaits != nil {
r.SeedWaits(p.Ask, true)
}
r.logf("the proposal %s is dropped by %s (%s); its ask %s is cancelled", key, by, why, p.Ask)
return "the ask " + p.Ask + " is cancelled and the proposal " + key + " is dropped", nil
}
// Heard takes a warrant the router said for the indexer (§667 Heard). It is applied only when its ask's id is a key's current
// ask, its digest the key's, and when the router's own record of the ask holds it with Warrant.For, never the event's body
// alone. A warrant of a replaced ask writes nothing.
func (r *Registry) Heard(ctx context.Context, w oc.Warrant) error {
r.mu.Lock()
defer r.mu.Unlock()
key, p, err := r.findAsk(ctx, w.Ask)
if err != nil {
return err
}
if p == nil || p.State != ProposalOpen {
r.logf("the warrant of the ask %s (%s) matches no open proposal, so it writes nothing", w.Ask, w.Outcome)
return nil
}
if w.Outcome != occ.OutcomeChosen {
state := ProposalRefused
switch w.Outcome {
case oc.WarrantOutcomeExpired:
state = ProposalExpired
case oc.WarrantOutcomeCancelled:
state = ProposalDropped
case oc.WarrantOutcomeReplaced:
r.logf("the warrant of the ask %s says it was replaced by a newer ask of %s", w.Ask, key)
return nil
}
return r.endKey(ctx, key, *p, state, "the ask "+w.Outcome+": "+w.Words)
}
// The router's own record of the ask holds the warrant (ADR 0286 §26, §27, §36).
state, rec, err := r.Where(p.Ask)
if err != nil {
return fmt.Errorf("the router's record of %s cannot be read, so its warrant is left unapplied: %w", p.Ask, err)
}
if rec == nil {
return fmt.Errorf("the router's record of %s (%q) holds no warrant, so the warrant heard is left unapplied", p.Ask, state)
}
if rec.Ask != w.Ask || rec.Option != w.Option || rec.AskDigest != w.AskDigest || rec.Asker != w.Asker {
return r.endKey(ctx, key, *p, ProposalRefused, "the warrant heard is not the one the router's record holds")
}
o, err := occ.For(*rec, Module, p.composeAsk(key))
if err == nil {
err = occ.Performs(o, p.act())
}
if err != nil {
return r.endKey(ctx, key, *p, ProposalRefused, err.Error())
}
if o.ID == optionDecline {
return r.endKey(ctx, key, *p, ProposalDeclined, "declined: "+occ.Says(*rec))
}
if p.Verb == verbSeed {
return r.applySeed(ctx, key, *p, *rec)
}
return r.apply(ctx, key, *p, *rec)
}
func (r *Registry) endKey(ctx context.Context, key string, p Proposal, state, outcome string) error {
if key == SeedKey && state != ProposalApplied && r.SeedWaits != nil {
state = ProposalSeedWaits
r.SeedWaits(p.Ask, true)
}
if err := r.end(ctx, key, p.Ask, state, outcome); err != nil {
return err
}
r.logf("the proposal %s ended %s: %s", key, state, outcome)
return nil
}
// findAsk is the key whose current ask has the id.
func (r *Registry) findAsk(ctx context.Context, ask string) (string, *Proposal, error) {
all, err := r.Proposals.Keys(ctx)
if err != nil {
return "", nil, err
}
for _, k := range all {
p, _, found, err := r.load(ctx, k)
if err != nil || !found {
continue
}
if p.Ask == ask {
return k, &p, nil
}
}
return "", nil, nil
}
// apply writes a warranted change: its history version first (kept once under <module>.<version>), then the registration by
// compare-and-set, then the event; the key ends applied. A change composed against a registration that moved since is refused.
func (r *Registry) apply(ctx context.Context, key string, p Proposal, w oc.Warrant) error {
cur, found, err := r.Registration(ctx, p.Module)
if err != nil {
return err
}
if cur.Version != p.Was || (!found && p.Was != 0) {
return r.endKey(ctx, key, p, ProposalRefused, fmt.Sprintf("the registration of %s moved to version %d since the change was asked at version %d; ask it again", p.Module, cur.Version, p.Was))
}
next, _, err := compose(p.Verb, cur, found, p.Change)
if err != nil {
return r.endKey(ctx, key, p, ProposalRefused, err.Error())
}
return r.write(ctx, key, p, next, w)
}
func (r *Registry) write(ctx context.Context, key string, p Proposal, next meshcatalog.Registration, w oc.Warrant) error {
at := r.now()
next.V, next.Version, next.Warrant, next.By, next.At = 1, p.Was+1, w.Ask, p.Caller, &at
if r.History != nil {
if err := r.History(ctx, next); err != nil {
return fmt.Errorf("the version %d of %s is not kept on %s, so it is not written: %w", next.Version, next.Module, RegistrationHistoryStream, err)
}
}
data, err := json.Marshal(next)
if err != nil {
return err
}
err = write(ctx, r.Registrations, RegistrationKey(next.Module), func(old []byte, found bool) ([]byte, bool, error) {
var cur meshcatalog.Registration
if found && json.Unmarshal(old, &cur) == nil && cur.Version != p.Was {
return nil, false, fmt.Errorf("the registration of %s is at version %d, not %d", next.Module, cur.Version, p.Was)
}
return data, false, nil
})
if err != nil {
return err
}
if r.Say != nil {
if err := r.Say(ctx, meshcatalog.RegistrationChangedEvent{V: 1, Module: next.Module, Version: next.Version, Moved: p.Lines, By: p.Caller, Warrant: w.Ask, At: &at}); err != nil {
r.logf("registration-changed of %s %d is not said: %v", next.Module, next.Version, err)
}
}
return r.endKey(ctx, key, p, ProposalApplied, fmt.Sprintf("applied as version %d of %s on %s", next.Version, next.Module, occ.Says(w)))
}
// Restart reads each open key at the indexer's start (§667 A restart): a decided ask is applied or ended, an open one waited on,
// and one the router no longer holds ended lost, said.
func (r *Registry) Restart(ctx context.Context) error {
all, err := r.Proposals.Keys(ctx)
if err != nil {
return err
}
slices.Sort(all)
var errs []error
for _, k := range all {
r.mu.Lock()
p, _, found, err := r.load(ctx, k)
r.mu.Unlock()
if err != nil || !found || p.State != ProposalOpen {
continue
}
state, w, err := r.Where(p.Ask)
switch {
case err != nil:
errs = append(errs, fmt.Errorf("%s: %w", k, err))
case state == "" && w == nil:
r.mu.Lock()
err := r.endKey(ctx, k, p, ProposalLost, "the router no longer holds the ask "+p.Ask)
r.mu.Unlock()
if err != nil {
errs = append(errs, err)
}
case w != nil:
if err := r.Heard(ctx, *w); err != nil {
errs = append(errs, err)
}
default:
r.logf("the proposal %s waits for the ask %s (%s)", k, p.Ask, state)
}
}
return errors.Join(errs...)
}
// RegistryAct is one act of the registrations asked through the takeins state (the way every tool of the indexer asks it):
// the verb (register, policy, watch, unwatch, registrations, drop or seed), its input as JSON, and for a drop the key.
type RegistryAct struct {
Verb string `json:"verb"`
Input json.RawMessage `json:"input,omitempty"`
Key string `json:"key,omitempty"`
Again bool `json:"again,omitempty"`
}
// The verbs of a RegistryAct that are no change of a registration.
const (
ActRegistrations = "registrations"
ActDrop = "drop"
ActSeed = "seed"
)
// NewRegistryAsk is the value and the key in takeins an act of the registrations is written under.
func NewRegistryAsk(act RegistryAct, why, by string, at time.Time) (string, []byte, error) {
if act.Verb == "" {
return "", nil, errors.New("name the act")
}
data, err := json.Marshal(TakeInAsk{V: 1, Registry: &act, Why: why, By: by, At: at.UTC()})
return "registry." + act.Verb + "." + at.UTC().Format("20060102T150405.000000000"), data, err
}
// Act answers one RegistryAct: the verb's answer as JSON in Registry, or why it could not be answered. seed is what the seed is
// composed from (the trunk's modules and what their build seats answered steps with).
func (r *Registry) Act(ctx context.Context, a TakeInAsk, seed func(context.Context) ([]SeedModule, error)) TakenIn {
out := TakenIn{Module: "registry"}
act := a.Registry
if act == nil {
out.Answer = "no registry act"
return out
}
var (
answer any
err error
)
switch act.Verb {
case verbRegister, verbPolicy, verbWatch, verbUnwatch:
answer, err = r.Propose(ctx, act.Verb, a.By, act.Input)
case ActRegistrations:
var in meshcatalog.RegistrationsInput
if len(act.Input) > 0 {
err = json.Unmarshal(act.Input, &in)
}
if err == nil {
answer, err = r.RegistrationsAnswer(ctx, in)
}
case ActDrop:
var said string
said, err = r.Drop(ctx, act.Key, a.Why, a.By)
answer = map[string]string{"answer": said}
case ActSeed:
var modules []SeedModule
if seed == nil {
err = errors.New("the seed is not composed by this indexer")
} else if modules, err = seed(ctx); err == nil {
var ask string
ask, err = r.ProposeSeed(ctx, modules, act.Again, a.By)
answer = map[string]string{"ask": ask}
}
default:
err = fmt.Errorf("%q is no act of the registrations", act.Verb)
}
if err != nil {
out.Answer = err.Error()
return out
}
out.Stored = true
if out.Registry, err = json.Marshal(answer); err != nil {
out.Stored, out.Answer = false, err.Error()
return out
}
out.Answer = act.Verb + " answered"
return out
}
// RegistryAct answers a registrations' act asked through takeins (design 47 G §3, §4, §8, §667).
func (ix *Indexer) RegistryAct(ctx context.Context, a TakeInAsk) TakenIn {
if ix.Registry == nil {
return TakenIn{Module: "registry", Answer: "this indexer keeps no registrations: their states are not bound yet"}
}
return ix.Registry.Act(ctx, a, ix.SeedModules)
}
@@ -0,0 +1,294 @@
package index
import (
"context"
"encoding/json"
"errors"
"strings"
"testing"
"time"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
gitseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/git"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// regRig is a Registry on memory states with a fake router: asks it was given, the cancelled ones, the history and the events.
type regRig struct {
r *Registry
asks map[string]oc.Ask
cancelled []string
history []meshcatalog.Registration
events []meshcatalog.RegistrationChangedEvent
now time.Time
unprotect map[string]bool
whereErr error
chosen map[string]string // ask id -> option the router's record holds
missing map[string]bool
}
func newRegRig() *regRig {
g := &regRig{asks: map[string]oc.Ask{}, now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC), unprotect: map[string]bool{}, chosen: map[string]string{}, missing: map[string]bool{}}
g.r = &Registry{Registrations: MemoryStore(), Proposals: MemoryStore(),
History: func(_ context.Context, r meshcatalog.Registration) error {
g.history = append(g.history, r)
return nil
},
Say: func(_ context.Context, e meshcatalog.RegistrationChangedEvent) error {
g.events = append(g.events, e)
return nil
},
Branch: func(repo, branch string) (gitseat.BranchAnswer, error) {
if branch == "" {
branch = "main"
}
return gitseat.BranchAnswer{Branch: branch, Repository: repo, Protected: !g.unprotect[branch]}, nil
},
Ask: func(a oc.Ask) error { g.asks[a.ID] = a; return nil },
Cancel: func(id string) error { g.cancelled = append(g.cancelled, id); return nil },
Where: func(id string) (string, *oc.Warrant, error) {
if g.whereErr != nil {
return "", nil, g.whereErr
}
if g.missing[id] {
return "", nil, nil
}
if opt, ok := g.chosen[id]; ok {
w := g.warrant(id, opt)
return "chosen", &w, nil
}
return "open", nil, nil
},
Now: func() time.Time { return g.now }, Logf: func(string, ...any) {}}
return g
}
// warrant is the warrant of the operator's choice of option on an ask, as the router would say it.
func (g *regRig) warrant(id, option string) oc.Warrant {
a := g.asks[id]
o, _ := occ.OptionOf(a, option)
return oc.Warrant{Ask: id, Asker: Module, Outcome: occ.OutcomeChosen, Option: option, Level: o.Level, Label: o.Label,
AskDigest: occ.Digest(a), At: occ.FormatTime(g.now.Add(time.Minute)), By: &oc.Person{Who: occ.Operator, Kind: "telegram", Verified: "user id verified"}}
}
func (g *regRig) decide(t *testing.T, ask, option string) {
t.Helper()
g.chosen[ask] = option
if err := g.r.Heard(context.Background(), g.warrant(ask, option)); err != nil {
t.Fatalf("hearing the warrant of %s: %v", ask, err)
}
}
func reg(module string, steps meshcatalog.StepStatuses, branches ...string) json.RawMessage {
b, _ := json.Marshal(meshcatalog.RegisterInput{Registration: meshcatalog.Registration{Module: module, Origin: "novox/" + module, Path: "module.json", Branches: branches, Policy: meshcatalog.Policy{Steps: steps}}})
return b
}
func pol(module string, s meshcatalog.StepStatuses) json.RawMessage {
b, _ := json.Marshal(meshcatalog.PolicyInput{Module: module, Steps: s})
return b
}
func TestRegistrationsAreWrittenOnTheirWarrantOnly(t *testing.T) {
ctx := context.Background()
g := newRegRig()
// Refusals raise no ask: a code review that is required (§3), a branch named candidate or entry (§343), an unprotected branch (§5).
g.unprotect["beta"] = true
for _, c := range []struct {
name, why string
change json.RawMessage
}{
{"review", meshcatalog.RegistrationAskedRefusedReviewRequired, reg("a", meshcatalog.StepStatuses{Review: "required"}, "main")},
{"unprotected", meshcatalog.RegistrationAskedRefusedUnprotectedBranch, reg("a", meshcatalog.StepStatuses{}, "beta")},
{"candidate", meshcatalog.RegistrationAskedRefusedUnprotectedBranch, reg("a", meshcatalog.StepStatuses{}, "candidate")},
} {
if c.name == "candidate" {
// the name is refused whatever the forge says of it
c.why = meshcatalog.RegistrationAskedRefusedUnprotectedBranch
}
got, err := g.r.Propose(ctx, "register", "session@node", c.change)
if err != nil || got.Refused == "" || got.Ask != "" || got.Words == "" {
t.Fatalf("%s: answer %+v, %v: want a refusal naming why and no ask", c.name, got, err)
}
}
if len(g.asks) != 0 {
t.Fatalf("a refusal raised %d asks", len(g.asks))
}
// A registration with no branch takes the default branch through the seat git (§342) and asks, naming each value and carrying the digest.
got, err := g.r.Propose(ctx, "register", "session@node", reg("a", meshcatalog.StepStatuses{Unit: "required", Lint: "bypass"}))
if err != nil || got.Ask == "" || len(got.Lines) < 3 {
t.Fatalf("register: %+v, %v", got, err)
}
ask := g.asks[got.Ask]
if !strings.Contains(ask.Explanation, "watching main") || !strings.Contains(ask.Explanation, "sha256:") || ask.About == "" {
t.Fatalf("the ask does not name the values and the digest: %+v", ask)
}
if all, _ := g.r.RegistrationsAnswer(ctx, meshcatalog.RegistrationsInput{}); len(all.Registrations) != 0 {
t.Fatalf("a registration was written before its warrant: %+v", all)
}
g.decide(t, got.Ask, "approve")
one, _ := g.r.RegistrationsAnswer(ctx, meshcatalog.RegistrationsInput{Module: "a"})
if len(one.Registrations) != 1 || one.Registrations[0].Version != 1 || one.Registrations[0].Warrant != got.Ask || one.Registrations[0].By != "session@node" {
t.Fatalf("after the warrant: %+v", one)
}
if len(g.history) != 1 || len(g.events) != 1 || g.events[0].Version != 1 || g.events[0].Warrant != got.Ask {
t.Fatalf("history %d, events %+v", len(g.history), g.events)
}
// A policy from required to bypass asks, showing that line; a decline writes nothing and ends the key declined.
p, err := g.r.Propose(ctx, "policy", "s@n", pol("a", meshcatalog.StepStatuses{Unit: "bypass"}))
if err != nil || len(p.Lines) != 1 || p.Lines[0] != "a unit tests: required -> bypass" {
t.Fatalf("policy: %+v, %v", p, err)
}
g.decide(t, p.Ask, "decline")
if now, _, _ := g.r.Registration(ctx, "a"); now.Version != 1 {
t.Fatalf("a declined change was written: %+v", now)
}
rows, _ := ListProposals(ctx, g.r.Proposals)
if len(rows) != 2 || rows[1].State != ProposalDeclined && rows[0].State != ProposalDeclined {
t.Fatalf("proposals: %+v", rows)
}
// Two changes of one key leave one open ask; the first ask's late warrant writes nothing.
g.now = g.now.Add(time.Hour)
first, _ := g.r.Propose(ctx, "policy", "s@n", pol("a", meshcatalog.StepStatuses{Unit: "bypass"}))
g.now = g.now.Add(time.Hour)
second, _ := g.r.Propose(ctx, "policy", "s@n", pol("a", meshcatalog.StepStatuses{Unit: "optional"}))
if first.Ask == second.Ask {
t.Fatal("a newer change of the key kept the older ask")
}
g.chosen[first.Ask] = "approve"
_ = g.r.Heard(ctx, g.warrant(first.Ask, "approve"))
if now, _, _ := g.r.Registration(ctx, "a"); now.Version != 1 {
t.Fatalf("the replaced ask's warrant wrote: %+v", now)
}
// Where the router's record cannot be read the warrant is left unapplied, said as an error.
g.whereErr = errors.New("no runtime")
if err := g.r.Heard(ctx, g.warrant(second.Ask, "approve")); err == nil {
t.Fatal("a warrant was applied with the router's record unread")
}
g.whereErr = nil
g.decide(t, second.Ask, "approve")
if now, _, _ := g.r.Registration(ctx, "a"); now.Version != 2 || now.Policy.Steps.Unit != "optional" {
t.Fatalf("the second change: %+v", now)
}
// watch adds a branch only when protected; unwatch removes it; each on its warrant.
w, _ := g.r.Propose(ctx, "watch", "s@n", mustJSON(meshcatalog.WatchInput{Module: "a", Branch: "beta"}))
if w.Refused != meshcatalog.RegistrationAskedRefusedUnprotectedBranch {
t.Fatalf("watch of an unprotected branch: %+v", w)
}
delete(g.unprotect, "beta")
w, _ = g.r.Propose(ctx, "watch", "s@n", mustJSON(meshcatalog.WatchInput{Module: "a", Branch: "beta"}))
g.decide(t, w.Ask, "approve")
u, _ := g.r.Propose(ctx, "unwatch", "s@n", mustJSON(meshcatalog.WatchInput{Module: "a", Branch: "beta"}))
g.decide(t, u.Ask, "approve")
if now, _, _ := g.r.Registration(ctx, "a"); now.Version != 4 || len(now.Branches) != 1 {
t.Fatalf("watch and unwatch: %+v", now)
}
if un, _ := g.r.Propose(ctx, "policy", "s@n", pol("zz", meshcatalog.StepStatuses{Unit: "bypass"})); un.Refused != meshcatalog.RegistrationAskedRefusedUnknownModule {
t.Fatalf("a policy of an unregistered module: %+v", un)
}
}
func mustJSON(v any) json.RawMessage { b, _ := json.Marshal(v); return b }
func TestProposalsSurviveARestartAndAreDropped(t *testing.T) {
ctx := context.Background()
g := newRegRig()
a, _ := g.r.Propose(ctx, "register", "s@n", reg("a", meshcatalog.StepStatuses{}, "main"))
b, _ := g.r.Propose(ctx, "register", "s@n", reg("b", meshcatalog.StepStatuses{}, "main"))
c, _ := g.r.Propose(ctx, "register", "s@n", reg("c", meshcatalog.StepStatuses{}, "main"))
// While the indexer was down the operator approved a and the router lost c; b is still open.
g.chosen[a.Ask] = "approve"
g.missing[c.Ask] = true
if err := g.r.Restart(ctx); err != nil {
t.Fatal(err)
}
rows, _ := ListProposals(ctx, g.r.Proposals)
states := map[string]string{}
for _, row := range rows {
states[row.Module] = row.State
}
if states["a"] != ProposalApplied || states["b"] != ProposalOpen || states["c"] != ProposalLost {
t.Fatalf("after a restart: %v", states)
}
// indexer_proposal_drop cancels the open ask and ends the key dropped; it asks why.
if _, err := g.r.Drop(ctx, ProposalKey("register", "b", ""), "", "s@n"); err == nil {
t.Fatal("a drop with no why was taken")
}
if _, err := g.r.Drop(ctx, ProposalKey("register", "b", ""), "not wanted", "s@n"); err != nil {
t.Fatal(err)
}
if len(g.cancelled) != 1 || g.cancelled[0] != b.Ask {
t.Fatalf("cancelled %v", g.cancelled)
}
rows, _ = ListProposals(ctx, g.r.Proposals)
for _, row := range rows {
if row.Module == "b" && row.State != ProposalDropped {
t.Fatalf("a dropped key: %+v", row)
}
}
}
func TestSeedIsOneAskAndWaitsForEveryBuildSeat(t *testing.T) {
ctx := context.Background()
g := newRegRig()
g.unprotect["main"] = false
mods := []SeedModule{{Module: "a", Origin: "novox/a", Path: "module.json", Artifacts: []SeedArtifact{{Name: "program", Seat: "node-build-go", Answered: false}}}}
if _, err := g.r.ProposeSeed(ctx, mods, false, "indexer"); err == nil || !strings.Contains(err.Error(), "node-build-go") {
t.Fatalf("an unanswered build seat did not hold the seed: %v", err)
}
mods[0].Artifacts[0].Answered, mods[0].Artifacts[0].Steps = true, []string{"unit"}
waits := ""
g.r.SeedWaits = func(ask string, waiting bool) {
if waiting {
waits = ask
} else {
waits = ""
}
}
ask, err := g.r.ProposeSeed(ctx, mods, false, "indexer")
if err != nil || ask == "" {
t.Fatalf("seed: %q, %v", ask, err)
}
// A step the seat answers none for is bypass; the statuses the check judged are the rest.
g.decide(t, ask, "decline")
if waits != ask {
t.Fatalf("a declined seed does not keep seed-waits: %q", waits)
}
if _, err := g.r.ProposeSeed(ctx, mods, false, "indexer"); err == nil {
t.Fatal("the seed was asked again with no indexer_seed_again")
}
g.now = g.now.Add(time.Hour)
ask, err = g.r.ProposeSeed(ctx, mods, true, "s@n")
if err != nil {
t.Fatal(err)
}
g.decide(t, ask, "approve")
got, _, _ := g.r.Registration(ctx, "a")
if got.Version != 1 || got.Policy.Steps.Review != "optional" || got.Policy.Steps.Approval != "bypass" || got.Policy.Artifacts["program"].Lint != "bypass" || got.Policy.Artifacts["program"].Integration != "bypass" || got.Policy.Artifacts["program"].Unit != "" {
t.Fatalf("the seeded registration: %+v", got)
}
if waits != "" {
t.Fatalf("seed-waits is not cleared by the warrant: %q", waits)
}
// An unprotected default branch is left out and named.
g2 := newRegRig()
g2.unprotect["main"] = true
seed, err := g2.r.ComposeSeed(mods)
if err != nil || len(seed.Registrations) != 0 || len(seed.Left) != 1 || !strings.Contains(seed.Left[0], "not protected") {
t.Fatalf("seed with an unprotected default branch: %+v, %v", seed, err)
}
}
func TestCanonicalIsRFC8785(t *testing.T) {
got, err := Canonical(map[string]any{"b": []any{1, "x
"}, "a": map[string]any{"z": true, "y": nil}, "é": "<&>"})
if err != nil {
t.Fatal(err)
}
if want := "{\"a\":{\"y\":null,\"z\":true},\"b\":[1,\"x
\"],\"é\":\"<&>\"}"; string(got) != want {
t.Fatalf("canonical = %s, want %s", got, want)
}
if _, err := Canonical(map[string]any{"n": 1.5}); err == nil {
t.Fatal("a fraction has no canonical form here")
}
}
@@ -0,0 +1,119 @@
package index
import (
"context"
"encoding/json"
"errors"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// RegistrationTools are the seat mesh-catalog's verbs registrations, register, policy, watch and unwatch (design 47 G §3, §4), and
// the indexer's own tools of the proposals (§667) and the seed (§8), served in the tools bundle. Every act is asked of the
// indexer through takeins, as the indexer's other tools are (ask writes the value and waits for the answer); the proposals
// are read from the state as the bundle reads it. record says a hand act in the hand-act log.
func RegistrationTools(proposals Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(cause, verb, what, why string) error) []stdio.Tool {
act := func(verb string, withInput bool) func(map[string]any) (any, error) {
return func(args map[string]any) (any, error) {
var input json.RawMessage
if withInput {
clean := map[string]any{}
for k, v := range args {
if k != "caller" && !strings.HasPrefix(k, "_") {
clean[k] = v
}
}
var err error
if input, err = json.Marshal(clean); err != nil {
return nil, err
}
}
why, _ := args["why"].(string)
key, value, err := NewRegistryAsk(RegistryAct{Verb: verb, Input: input}, why, stdio.CallerOf(args), now())
if err != nil {
return nil, err
}
return answerOf(ask(key, value))
}
}
return []stdio.Tool{
{Name: meshcatalog.Seat + "." + meshcatalog.VerbRegistrations, Run: act(ActRegistrations, true)},
{Name: meshcatalog.Seat + "." + meshcatalog.VerbRegister, Run: act(verbRegister, true)},
{Name: meshcatalog.Seat + "." + meshcatalog.VerbPolicy, Run: act(verbPolicy, true)},
{Name: meshcatalog.Seat + "." + meshcatalog.VerbWatch, Run: act(verbWatch, true)},
{Name: meshcatalog.Seat + "." + meshcatalog.VerbUnwatch, Run: act(verbUnwatch, true)},
{Name: ProposalsTool, Description: "The changes of the registrations that wait for the operator's warrant and those that ended (design 47 G §667): " +
"for each key (<verb>.<module>, with the branch for watch and unwatch, and seed) its state (open, applied, declined, " +
"expired, refused, dropped, lost or seed-waits), its ask, caller, the lines it moves, its digest and its outcome. An ended key stays until the next change of the same key replaces it.",
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) {
rows, err := ListProposals(context.Background(), proposals)
return map[string]any{"proposals": rows}, err
}},
{Name: ProposalDropTool, Description: "Cancel the open ask of one key of the proposals and end the key dropped (design 47 G §667). A hand act: who and why " +
"are said in the hand-act log. indexer_proposals names the keys.",
Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{
"key": map[string]any{"type": "string", "description": "the proposal's key, as indexer_proposals names it"},
"why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}},
Run: func(args map[string]any) (any, error) {
key, _ := args["key"].(string)
why, _ := args["why"].(string)
if strings.TrimSpace(key) == "" || strings.TrimSpace(why) == "" {
return nil, errors.New("name the key (key) and say why (why): a hand act is said in the hand-act log")
}
k, value, err := NewRegistryAsk(RegistryAct{Verb: ActDrop, Key: strings.TrimSpace(key)}, why, stdio.CallerOf(args), now())
if err != nil {
return nil, err
}
out, err := answerOf(ask(k, value))
if err != nil {
return out, err
}
res := map[string]any{"result": out}
if rerr := record(ProposalDropTool, "proposal-drop", "dropped the proposal "+key, why); rerr != nil {
res["hand-act"] = "not recorded: " + rerr.Error()
}
return res, nil
}},
{Name: SeedAgainTool, Description: "Raise the ask of the first registrations again from the seed's key (design 47 G §8), after a Decline, an expiry or a drop " +
"ended it and while the indexer holds no registration. A hand act: who and why are said in the hand-act log.",
Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{
"why": map[string]any{"type": "string", "description": "why it is asked again, in plain words"}}},
Run: func(args map[string]any) (any, error) {
why, _ := args["why"].(string)
if strings.TrimSpace(why) == "" {
return nil, errors.New("say why (why): a hand act is said in the hand-act log")
}
k, value, err := NewRegistryAsk(RegistryAct{Verb: ActSeed, Again: true}, why, stdio.CallerOf(args), now())
if err != nil {
return nil, err
}
out, err := answerOf(ask(k, value))
if err != nil {
return out, err
}
res := map[string]any{"result": out}
if rerr := record(SeedAgainTool, "seed-again", "asked the first registrations again", why); rerr != nil {
res["hand-act"] = "not recorded: " + rerr.Error()
}
return res, nil
}},
}
}
// answerOf is the answer of an act the indexer answered: the verb's own answer, or the words that say why not.
func answerOf(a TakeInAsk, err error) (any, error) {
if err != nil {
return nil, err
}
if a.Result == nil {
return nil, errors.New("the indexer gave no answer")
}
if !a.Result.Stored {
return nil, errors.New(a.Result.Answer)
}
return a.Result.Registry, nil
}
@@ -0,0 +1,171 @@
package index
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"strings"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// SeedModule is one module the trunk revision holds, as the seed reads it (design 47 G §8): its origin and the path of its
// manifest, and for each artifact whether its build seat answered `steps` and with which steps.
type SeedModule struct {
Module string
Origin string
Path string
Artifacts []SeedArtifact
}
// SeedArtifact is an artifact of a module and what its build seat answered `steps` with: Answered false is a seat that has not
// answered, which holds the seed (never read as none), and Steps empty with Answered true is none, so every step the
// artifact's seat detects none for is bypass.
type SeedArtifact struct {
Name string
Seat string
Answered bool
Steps []string
}
// SeedProposal is the first registrations composed from what the indexer took in (§8): the registrations proposed, the
// modules left out because their default branch is not protected, and the build seats still unanswered.
type SeedProposal struct {
Registrations []meshcatalog.Registration
Left []string
Unanswered []string
}
// ComposeSeed proposes one registration for every module its trunk revision holds (§8): the origin and path of each, its
// watched branch the repository's default branch read through the seat git, and the statuses the check judged until then:
// lint, unit tests, integration tests and composition required, code review optional, approval bypass; each step an
// artifact's build seat answers steps with none is bypass, so no module is held by a required step it has nothing for. A
// module whose default branch the forge does not protect is left out and named. A build seat that has not answered holds
// the seed.
func (r *Registry) ComposeSeed(modules []SeedModule) (SeedProposal, error) {
var out SeedProposal
slices.SortFunc(modules, func(a, b SeedModule) int { return strings.Compare(a.Module, b.Module) })
for _, m := range modules {
for _, a := range m.Artifacts {
if !a.Answered {
out.Unanswered = append(out.Unanswered, a.Seat)
}
}
}
slices.Sort(out.Unanswered)
out.Unanswered = slices.Compact(out.Unanswered)
if len(out.Unanswered) > 0 {
return out, nil
}
for _, m := range modules {
b, err := r.Branch(m.Origin, "")
if err != nil {
return out, fmt.Errorf("the seat git does not answer the default branch of %s: %w", m.Origin, err)
}
if b.Error != "" || b.Branch == "" {
return out, fmt.Errorf("the default branch of %s is not read: %s", m.Origin, b.Error)
}
if p, err := r.Branch(m.Origin, b.Branch); err != nil || !p.Protected || p.Error != "" {
out.Left = append(out.Left, fmt.Sprintf("%s (the branch %s of %s is not protected)", m.Module, b.Branch, m.Origin))
continue
}
reg := meshcatalog.Registration{Module: m.Module, Origin: m.Origin, Path: m.Path, Branches: []string{b.Branch},
Policy: meshcatalog.Policy{Steps: meshcatalog.StepStatuses{Lint: "required", Unit: "required", Integration: "required",
Composition: "required", Review: "optional", Approval: "bypass"}}}
for _, a := range m.Artifacts {
var s meshcatalog.StepStatuses
if !slices.Contains(a.Steps, "lint") {
s.Lint = "bypass"
}
if !slices.Contains(a.Steps, "unit") {
s.Unit = "bypass"
}
if !slices.Contains(a.Steps, "integration") {
s.Integration = "bypass"
}
if s != (meshcatalog.StepStatuses{}) {
if reg.Policy.Artifacts == nil {
reg.Policy.Artifacts = map[string]meshcatalog.StepStatuses{}
}
reg.Policy.Artifacts[a.Name] = s
}
}
out.Registrations = append(out.Registrations, reg)
}
return out, nil
}
// ProposeSeed raises the first registrations as one ask (§8), when the indexer holds no registration and the seed's key is not
// open or applied; again (indexer_seed_again) it raises the ask once more from a key a Decline, an expiry or a drop ended.
// It answers the ask's id, or why none was raised.
func (r *Registry) ProposeSeed(ctx context.Context, modules []SeedModule, again bool, by string) (string, error) {
r.mu.Lock()
defer r.mu.Unlock()
held, err := r.Registrations.Keys(ctx)
if err != nil {
return "", err
}
if len(held) > 0 {
return "", errors.New("the indexer holds registrations already: the seed is for the first start only")
}
if old, _, found, err := r.load(ctx, SeedKey); err != nil {
return "", err
} else if found && old.State == ProposalOpen {
return old.Ask, nil
} else if found && !again {
return "", fmt.Errorf("the seed ended %s: indexer_seed_again asks it again", old.State)
}
seed, err := r.ComposeSeed(modules)
if err != nil {
return "", err
}
if len(seed.Unanswered) > 0 {
return "", fmt.Errorf("the build seats %v have not answered steps: the seed waits for them (call-unanswered after five minutes)", seed.Unanswered)
}
if len(seed.Registrations) == 0 {
return "", errors.New("no module of the trunk revision can be registered: " + strings.Join(seed.Left, "; "))
}
change, err := json.Marshal(seed.Registrations)
if err != nil {
return "", err
}
p := Proposal{V: 1, Verb: verbSeed, Change: change, Seed: seed.Registrations, Caller: by, At: r.now(), State: ProposalOpen}
for _, g := range seed.Registrations {
p.Lines = append(p.Lines, fmt.Sprintf("%s: registered from %s at %s, watching %s", g.Module, g.Origin, g.Path, strings.Join(g.Branches, ", ")))
}
for _, l := range seed.Left {
p.Lines = append(p.Lines, "left out: "+l)
}
if p.Digest, err = DigestOf(map[string]any{"verb": verbSeed, "change": json.RawMessage(change)}); err != nil {
return "", err
}
if err := r.keepAndAsk(ctx, SeedKey, &p); err != nil {
return "", err
}
if r.SeedWaits != nil {
r.SeedWaits(p.Ask, false)
}
return p.Ask, nil
}
// applySeed writes every registration the seed proposes, each as version 1 on the warrant, and ends the key applied.
func (r *Registry) applySeed(ctx context.Context, key string, p Proposal, w oc.Warrant) error {
if held, err := r.Registrations.Keys(ctx); err != nil || len(held) > 0 {
return r.endKey(ctx, key, p, ProposalRefused, "the indexer holds registrations already, so the seed writes none")
}
for _, reg := range p.Seed {
q := p
q.Verb, q.Module, q.Was = verbRegister, reg.Module, 0
q.Lines = []string{fmt.Sprintf("%s: registered by the first registrations", reg.Module)}
if err := r.write(ctx, "", q, reg, w); err != nil {
return err
}
}
if r.SeedWaits != nil {
r.SeedWaits(p.Ask, false)
}
return r.end(ctx, key, p.Ask, ProposalApplied, fmt.Sprintf("applied: %d registrations on %s", len(p.Seed), w.Ask))
}
@@ -67,13 +67,16 @@ type TakeInAsk struct {
// Recompute asks the catalog graph to be computed again now and compared with the newest kept revision (indexer_recompute).
Recompute bool `json:"recompute,omitempty"`
// CandidateGroup and CandidateHeads, with Drop, ask that candidate's catalog half to be dropped (indexer_drop_candidate_graph).
CandidateGroup string `json:"candidate-group,omitempty"`
CandidateHeads string `json:"candidate-heads,omitempty"`
Drop bool `json:"drop,omitempty"`
By string `json:"by,omitempty"`
At time.Time `json:"at"`
Result *TakenIn `json:"result,omitempty"`
Answered time.Time `json:"answered,omitzero"`
CandidateGroup string `json:"candidate-group,omitempty"`
CandidateHeads string `json:"candidate-heads,omitempty"`
Drop bool `json:"drop,omitempty"`
// Registry asks one of the registrations' acts (registrations.go): a trusted change of one, the verb registrations,
// indexer_proposal_drop or indexer_seed_again (design 47 G §3, §4, §8, §667).
Registry *RegistryAct `json:"registry,omitempty"`
By string `json:"by,omitempty"`
At time.Time `json:"at"`
Result *TakenIn `json:"result,omitempty"`
Answered time.Time `json:"answered,omitzero"`
}
// Done says the ask is answered.
@@ -88,7 +91,7 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) {
if a.V != 1 {
return a, fmt.Errorf("a takeins value of version %d, not 1", a.V)
}
if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" {
if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.Registry == nil {
return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try")
}
return a, nil
+15 -10
View File
@@ -49,6 +49,7 @@ type stream struct {
Events []string `json:"events"`
MaxAgeDays int `json:"max-age-days"`
MaxMiB int `json:"max-mib"`
Retention string `json:"retention"`
}
type manifest struct {
@@ -122,7 +123,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: "own-build", History: 1}, {Name: index.PackagesState, History: index.PackagesHistory},
{Name: views.MarksState, History: 1}, {Name: mcc.FindingsState, History: 5}, {Name: views.StatusState, History: 1},
{Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5},
{Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}, {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.<state>` among its emits, and it has no stream of its own.
// ADR 0635 §5: its views family is its ordinary events, `views.<state>` among its emits, and its one stream is the
// registrations' history (design 47 G §2, asserted above).
// ADR 0636 §4: seat-definition-changed and -removed are the seat mesh-catalog's events, said on the seat's subject and kept
// on the shared events stream, and not among its own emits.
if len(m.Streams) != 0 {
t.Errorf("streams = %+v, want none", m.Streams)
}
for _, state := range []string{"candidates", "config-candidates", "findings", "graph", "heard", "manifests", "packages", "predefined", "seat-definitions", "taken", "verdicts"} {
if !slices.Contains(m.Emits, views.EventPrefix+state) {
t.Errorf("emits = %v: want %s%s", m.Emits, views.EventPrefix, state)
@@ -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} {
+18 -3
View File
@@ -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",