Merge branch 'feat/47-G-indexer6a' into feat/47-G-join8
# Conflicts: # contracts/seats/mesh-wants/seat.json
This commit is contained in:
@@ -948,10 +948,12 @@ type RegistrationAsked struct {
|
||||
|
||||
// The values of RegistrationAsked's refused.
|
||||
const (
|
||||
RegistrationAskedRefusedUnprotectedBranch = "unprotected branch"
|
||||
RegistrationAskedRefusedReviewRequired = "review required"
|
||||
RegistrationAskedRefusedUnknownModule = "unknown module"
|
||||
RegistrationAskedRefusedNothingMoves = "nothing moves"
|
||||
RegistrationAskedRefusedUnprotectedBranch = "unprotected branch"
|
||||
RegistrationAskedRefusedReservedBranchName = "reserved branch name"
|
||||
RegistrationAskedRefusedPlaced = "placed"
|
||||
RegistrationAskedRefusedReviewRequired = "review required"
|
||||
RegistrationAskedRefusedUnknownModule = "unknown module"
|
||||
RegistrationAskedRefusedNothingMoves = "nothing moves"
|
||||
)
|
||||
|
||||
// StepStatuses is the status of each pipeline step (design 47 *Pipeline steps and the policy*); a step left out keeps the module's.
|
||||
@@ -1427,6 +1429,8 @@ type ManifestAcceptedEvent struct {
|
||||
Digest string `json:"digest"`
|
||||
Judge string `json:"judge"`
|
||||
Module string `json:"module"`
|
||||
// each problem the indexer's re-judgement finds in a manifest it keeps accepted because its digest is unchanged: the manifest validator's strings, the same items its verdict writes as warnings, in the same order; an empty list when none (design 47 G section 668, 692)
|
||||
Problems []string `json:"problems"`
|
||||
// the indexer's revision of the verdict it says; a reader asks its first view again on a gap in it; zero from an indexer that does not say it
|
||||
Revision int64 `json:"revision,omitempty"`
|
||||
V int64 `json:"v"`
|
||||
|
||||
@@ -151,6 +151,10 @@ func TestFixtures(t *testing.T) {
|
||||
decode: func(b []byte) (any, error) { var v RegisterInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
|
||||
{name: "register refused for an unprotected branch", subject: "", given: "{\"module\":\"nats\",\"refused\":\"unprotected branch\",\"words\":\"the branch beta of novox/mesh-catalog admits a direct push\"}", wire: "{\"module\":\"nats\",\"refused\":\"unprotected branch\",\"words\":\"the branch beta of novox/mesh-catalog admits a direct push\"}",
|
||||
decode: func(b []byte) (any, error) { var v RegisterAnswer; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
|
||||
{name: "unwatch refused for a branch a placement names", subject: "", given: "{\"module\":\"nats\",\"refused\":\"placed\",\"words\":\"the branch beta of nats is placed: n1 and n2 run it from that branch\"}", wire: "{\"module\":\"nats\",\"refused\":\"placed\",\"words\":\"the branch beta of nats is placed: n1 and n2 run it from that branch\"}",
|
||||
decode: func(b []byte) (any, error) { var v UnwatchAnswer; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
|
||||
{name: "register refused for a reserved branch name", subject: "", given: "{\"module\":\"nats\",\"refused\":\"reserved branch name\",\"words\":\"the branch candidate of novox/mesh-catalog is a name the catalog keeps for its own revisions\"}", wire: "{\"module\":\"nats\",\"refused\":\"reserved branch name\",\"words\":\"the branch candidate of novox/mesh-catalog is a name the catalog keeps for its own revisions\"}",
|
||||
decode: func(b []byte) (any, error) { var v RegisterAnswer; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
|
||||
{name: "policy moves one artifact's unit step", subject: "", given: "{\"artifact\":\"program\",\"module\":\"nats\",\"steps\":{\"unit\":\"bypass\"},\"why\":\"its tests need hardware\"}", wire: "{\"artifact\":\"program\",\"module\":\"nats\",\"steps\":{\"unit\":\"bypass\"},\"why\":\"its tests need hardware\"}",
|
||||
decode: func(b []byte) (any, error) { var v PolicyInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
|
||||
{name: "policy asked, naming the moved status", subject: "", given: "{\"ask\":\"ask-m2kx9a1b3c4d5e\",\"lines\":[\"nats:program unit: required -\\u003e bypass\"],\"module\":\"nats\"}", wire: "{\"ask\":\"ask-m2kx9a1b3c4d5e\",\"lines\":[\"nats:program unit: required -\\u003e bypass\"],\"module\":\"nats\"}",
|
||||
|
||||
@@ -296,6 +296,24 @@
|
||||
"words": "the branch beta of novox/mesh-catalog admits a direct push"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "unwatch refused for a branch a placement names",
|
||||
"of": "verb:unwatch.answer",
|
||||
"given": {
|
||||
"module": "nats",
|
||||
"refused": "placed",
|
||||
"words": "the branch beta of nats is placed: n1 and n2 run it from that branch"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "register refused for a reserved branch name",
|
||||
"of": "verb:register.answer",
|
||||
"given": {
|
||||
"module": "nats",
|
||||
"refused": "reserved branch name",
|
||||
"words": "the branch candidate of novox/mesh-catalog is a name the catalog keeps for its own revisions"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "policy moves one artifact's unit step",
|
||||
"of": "verb:policy.input",
|
||||
|
||||
@@ -678,6 +678,13 @@
|
||||
"type": "integer",
|
||||
"description": "the indexer's revision of the verdict it says; a reader asks its first view again on a gap in it; zero from an indexer that does not say it"
|
||||
},
|
||||
"problems": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"description": "each problem the indexer's re-judgement finds in a manifest it keeps accepted because its digest is unchanged: the manifest validator's strings, the same items its verdict writes as warnings, in the same order; an empty list when none (design 47 G section 668, 692)"
|
||||
},
|
||||
"bus-state": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
@@ -717,7 +724,8 @@
|
||||
"module",
|
||||
"digest",
|
||||
"judge",
|
||||
"bus-state"
|
||||
"bus-state",
|
||||
"problems"
|
||||
]
|
||||
}
|
||||
},
|
||||
@@ -2424,6 +2432,8 @@
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"unprotected branch",
|
||||
"reserved branch name",
|
||||
"placed",
|
||||
"review required",
|
||||
"unknown module",
|
||||
"nothing moves"
|
||||
|
||||
+3
-1
@@ -448,7 +448,7 @@ export interface RegistrationAsked {
|
||||
"lines"?: string[];
|
||||
"module": string;
|
||||
/** why no ask was raised */
|
||||
"refused"?: "unprotected branch" | "review required" | "unknown module" | "nothing moves";
|
||||
"refused"?: "unprotected branch" | "reserved branch name" | "placed" | "review required" | "unknown module" | "nothing moves";
|
||||
/** the refusal in words, naming the branch, the step or the module */
|
||||
"words"?: string;
|
||||
}
|
||||
@@ -878,6 +878,8 @@ export interface ManifestAcceptedEvent {
|
||||
"digest": string;
|
||||
"judge": string;
|
||||
"module": string;
|
||||
/** each problem the indexer's re-judgement finds in a manifest it keeps accepted because its digest is unchanged: the manifest validator's strings, the same items its verdict writes as warnings, in the same order; an empty list when none (design 47 G section 668, 692) */
|
||||
"problems": string[];
|
||||
/** the indexer's revision of the verdict it says; a reader asks its first view again on a gap in it; zero from an indexer that does not say it */
|
||||
"revision"?: number;
|
||||
"v": number;
|
||||
|
||||
@@ -288,11 +288,15 @@ func main() {
|
||||
tools = append(tools, catalogTools(trunk.State)...)
|
||||
// The seat mesh-catalog's verb revision, a first view of the catalog half answered in this runtime of its own to exactly
|
||||
// the parts whose entry declares it (novox/hq ADR 0663).
|
||||
tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}))
|
||||
tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}, keptRead{stdio.State(index.NamedState)}))
|
||||
tools = append(tools, index.CatalogRevisionTools(revisions, asks, time.Now, func(what, why string) error {
|
||||
_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.DropCandidateGraphTool, Verb: "drop-candidate-graph"})
|
||||
return err
|
||||
})...)
|
||||
tools = append(tools, index.NamedTools(asks, time.Now, func(what, why string) error {
|
||||
_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.NamedDropTool, Verb: "named-drop"})
|
||||
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
|
||||
|
||||
@@ -29,11 +29,13 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
|
||||
@@ -595,6 +597,20 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, "FORGE", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The two seat events that name a fingerprint, each on its stream kept last per subject with the indexer's durable on it, and the
|
||||
// state named the grants part makes from the manifest (design 47 G §675, §695); both streams empty, so the start's read ends at once.
|
||||
for stream, subject := range map[string]string{meshwants.StreamWanted: meshwants.SubjectWantedAll, nodeengine.StreamRunning: nodeengine.SubjectRunningAll} {
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy,
|
||||
DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy, FilterSubjects: []string{subject}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.NamedBucket, History: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The build modules' streams for toolchain.reported (declared by them, heard by the indexer) and its durable consumers.
|
||||
for module, stream := range map[string]string{"mesh-build-go": "TOOLCHAINS_GO", "mesh-build-node": "TOOLCHAINS_NODE"} {
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.stream." + module + ".event.toolchain.reported"}}); err != nil {
|
||||
@@ -640,7 +656,9 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
|
||||
}
|
||||
t.Cleanup(func() { _ = waiter.Unsubscribe() })
|
||||
// The shared events stream: the indexer's ordinary events and the assigner's views events (ADR 0635 §5).
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil {
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.mesh-catalog.event.>", "mesh.seat.mesh-delivery.event.>", "mesh.seat.mesh-delivery-checker.event.>",
|
||||
"mesh.seat.mesh-conditions.event.>", "mesh.seat.operator-channel.event.>", "mesh.seat.go-package-registry.event.>", "mesh.seat.npm-package-registry.event.>",
|
||||
"mesh.seat.mesh-module-assigner.event.>"}}); err != nil { // not the seats kept in streams of their own (wanted and running, above)
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The indexer's own stream of its findings' events (design 47 G §383), which its finding writer says into (A-30).
|
||||
@@ -1414,3 +1432,45 @@ func serveAssignerAssignments(t *testing.T, ctx context.Context, js jetstream.Je
|
||||
}
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
}
|
||||
|
||||
// TestIndexerKeepsWhatAWantAndARunningBuildNameAndLetsAGoneOneLeave (design 47 G §675, §693, A-61): a want said before the start on
|
||||
// wanted.<node>.<module> and a running build of another node are read whole at the start into the state named; a later
|
||||
// wanted that says the want removed empties its member whatever the node last said running.
|
||||
func TestIndexerKeepsWhatAWantAndARunningBuildNameAndLetsAGoneOneLeave(t *testing.T) {
|
||||
fp := "src2:" + strings.Repeat("a", 64)
|
||||
at := time.Now().UTC()
|
||||
js, ctx, _, _ := startIndexer(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) {
|
||||
want, _ := json.Marshal(meshwants.WantedEvent{V: 1, Node: "n1", Module: "avahi", Fp: fp, Revision: 1, At: &at})
|
||||
run, _ := json.Marshal(nodeengine.Running{V: 1, Node: "n2", Module: "avahi", Fingerprint: fp, At: at})
|
||||
for subject, data := range map[string][]byte{meshwants.SubjectWanted("n1", "avahi"): want, nodeengine.SubjectRunning("n2", "avahi"): run} {
|
||||
if _, err := js.Publish(ctx, subject, data); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
})
|
||||
named, err := js.KeyValue(ctx, index.NamedBucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
members := func(want int) []string {
|
||||
var got []string
|
||||
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
got, _ = index.NamedMembers(ctx, kvStore{named}, fp)
|
||||
if len(got) == want {
|
||||
break
|
||||
}
|
||||
}
|
||||
return got
|
||||
}
|
||||
if got := members(2); !slices.Equal(got, []string{"n1.avahi", "n2.avahi"}) {
|
||||
t.Fatalf("named holds %v for the fingerprint within 20 s of the start, want both members", got)
|
||||
}
|
||||
gone := at.Add(time.Minute)
|
||||
removed, _ := json.Marshal(meshwants.WantedEvent{V: 1, Node: "n2", Module: "avahi", Fp: fp, Revision: 2, Removed: true, At: &gone})
|
||||
if _, err := js.Publish(ctx, meshwants.SubjectWanted("n2", "avahi"), removed); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := members(1); !slices.Equal(got, []string{"n1.avahi"}) {
|
||||
t.Errorf("named holds %v after the want of n2 was said removed, want n1.avahi alone", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -430,7 +430,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
})
|
||||
readWhenBound(seatReader, seatDefs, marks)
|
||||
ix := &index.Indexer{Findings: writer, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known,
|
||||
// Kept while named (design 47 G §675): the want and the running build that name a fingerprint, heard below.
|
||||
named := index.NewNamed(awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.NamedBucket, "no entry is kept for a want or a running build that names it", logger.Printf)}, logger.Printf)
|
||||
ix := &index.Indexer{NamedSets: named, Findings: writer, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known,
|
||||
Verdicts: awaitedStore{b: verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred),
|
||||
Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf,
|
||||
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
|
||||
@@ -450,8 +452,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
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)
|
||||
routerRecordOf(cred.Module, calls.Call, calls.Ask), ix, logger.Printf)
|
||||
ix.Registry = registry
|
||||
named.Changed = ix.RemoveUnlisted
|
||||
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)")
|
||||
}
|
||||
@@ -683,6 +686,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
go func() { <-ctx.Done(); hub.Stop() }()
|
||||
return nil
|
||||
})
|
||||
hearNamed(ctx, js, cred, notHeard, named, logger.Printf)
|
||||
// The checker's verdicts carry a pull request's out-of-step files (DLV-I20): on the indexer's durable consumer of the
|
||||
// checker's stream asks. Where it is not there yet, no pull request's out-of-step files are kept until the indexer
|
||||
// starts again, said.
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// binder binds an observer's consumer of a family and says in a finding when it cannot (heard.Writer).
|
||||
type binder interface {
|
||||
Bind(family, consequence string, bind func() error) error
|
||||
}
|
||||
|
||||
// hearNamed hears the two events that name a fingerprint (design 47 G §675, §693; A-61): the seat mesh-wants' `wanted.<node>.<module>`
|
||||
// and the node-engine seat's `running.<node>.<module>`, each on its own stream kept last per subject, read whole at the start
|
||||
// and then followed on the indexer's durable (§556). The end of each whole read is said to named, which reconciles its sets once
|
||||
// both are read; a stream the bus does not hold yet is the finding family-not-heard, and no entry is removed meanwhile.
|
||||
func hearNamed(ctx context.Context, js jetstream.JetStream, cred process.Account, notHeard binder, named *index.Named, logf func(string, ...any)) {
|
||||
for _, s := range []struct {
|
||||
stream, kind string
|
||||
each func(context.Context, []byte) error
|
||||
}{
|
||||
{meshwants.StreamWanted, index.NamedWanted, named.HearWanted},
|
||||
{nodeengine.StreamRunning, index.NamedRunning, named.HearRunning},
|
||||
} {
|
||||
_ = notHeard.Bind(s.stream, "no entry is removed, since what a want or a running build names is not heard until the indexer starts again", func() error {
|
||||
cc, err := cred.HearSeatEvents(ctx, js, s.stream, func(m jetstream.Msg) {
|
||||
if err := m.Ack(); err != nil {
|
||||
logf("acknowledging %s: %v", m.Subject(), err)
|
||||
}
|
||||
if err := s.each(ctx, m.Data()); err != nil {
|
||||
logf("%s: %v", m.Subject(), err)
|
||||
}
|
||||
}, logf, process.OnRead(func(state string) {
|
||||
if state != process.ReadDone {
|
||||
return
|
||||
}
|
||||
if err := named.ReadDone(ctx, s.kind); err != nil {
|
||||
logf("the sets of %s are not reconciled with %s: %v", index.NamedState, s.stream, err)
|
||||
}
|
||||
}))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
go func() { <-ctx.Done(); cc.Stop() }()
|
||||
return nil
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -3,14 +3,16 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
"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"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -47,8 +49,8 @@ func historySubject(module string) string {
|
||||
}
|
||||
|
||||
// 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 {
|
||||
// operator-channel's accepts. where is the read of the router's own record of an ask (routerRecordOf, §700).
|
||||
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), ix *index.Indexer, 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)
|
||||
@@ -56,6 +58,14 @@ func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account,
|
||||
return err
|
||||
}
|
||||
_, err = js.Publish(ctx, historySubject(r.Module), data, jetstream.WithMsgID(fmt.Sprintf("%s.%d", r.Module, r.Version)))
|
||||
if err == nil {
|
||||
// Past 80 % full the stream's fill is a finding (design 47 G §704).
|
||||
if st, serr := js.Stream(ctx, catalogue.StreamBusName(index.RegistrationHistoryStream)); serr == nil {
|
||||
if info, ierr := st.Info(ctx); ierr == nil {
|
||||
ix.HistoryFill(ctx, int64(info.State.Bytes), info.Config.MaxBytes)
|
||||
}
|
||||
}
|
||||
}
|
||||
return err
|
||||
},
|
||||
HistoryOf: func(ctx context.Context, module string) ([]meshcatalog.Registration, error) {
|
||||
@@ -88,6 +98,17 @@ func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account,
|
||||
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)
|
||||
},
|
||||
Placement: func(module string) ([]mma.ModulePlacement, error) {
|
||||
raw, err := call("seat:"+mma.Seat+"."+mma.VerbPlacement, mma.PlacementInput{Module: module})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var a mma.PlacementAnswer
|
||||
if err := json.Unmarshal(raw, &a); err != nil {
|
||||
return nil, fmt.Errorf("the answer of %s.%s is not its seat's: %w", mma.Seat, mma.VerbPlacement, err)
|
||||
}
|
||||
return a.Placements, nil
|
||||
},
|
||||
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})
|
||||
@@ -107,17 +128,49 @@ func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account,
|
||||
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("seed-waits is cleared: the first registrations are written (ask %s)", ask)
|
||||
}
|
||||
ix.SeedWaits(context.Background(), ask, waiting)
|
||||
},
|
||||
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)")
|
||||
// routerBucket is the bucket of the router's asks: the state `asks` of the seat operator-channel (the contract's StateAsks) in
|
||||
// the bucket of the seat's holder on the seat list, which the seat mesh-module-assigner's verb assignments answers with seats
|
||||
// (design 47 G §700, A-74); no module's name is in the program.
|
||||
func routerBucket(call toolCaller) (string, error) {
|
||||
raw, err := call("seat:"+mma.Seat+"."+mma.VerbAssignments, mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("who holds the seat %s is not asked of the seat %s: %w", oc.Seat, mma.Seat, err)
|
||||
}
|
||||
var held struct {
|
||||
Seats []mma.SeatHolders `json:"seats"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &held); err != nil {
|
||||
return "", fmt.Errorf("the answer of %s.%s is not its seat's: %w", mma.Seat, mma.VerbAssignments, err)
|
||||
}
|
||||
for _, s := range held.Seats {
|
||||
if s.Seat == oc.Seat && len(s.Holders) > 0 {
|
||||
return s.Holders[0].Module + "_" + oc.StateAsks, nil
|
||||
}
|
||||
}
|
||||
return "", fmt.Errorf("no module holds the seat %s on the seat list, so the router's record of an ask is not read", oc.Seat)
|
||||
}
|
||||
|
||||
func routerRecordUnread(string) (string, *oc.Warrant, error) { return "", nil, errRouterRecord }
|
||||
// routerRecordOf is Registry.Where: the router's own record of an ask, read under the indexer's module's name through
|
||||
// operatorchannelclient, whose Record is a get of that key (`<module>.<id>` of the seat's state asks) on the process's own bus
|
||||
// connection (record is process.Conn.Ask), as the tool runner gives it the runtime's mesh/seat.record (design 47 G §700): never
|
||||
// the decided event's body. A warrant is then left unapplied and said while the record cannot be read.
|
||||
func routerRecordOf(module string, call toolCaller, record func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (string, *oc.Warrant, error) {
|
||||
client := &operatorchannelclient.Client{Module: module, Record: func(id string) (*stdio.StateEntry, error) {
|
||||
bucket, err := routerBucket(call)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return record(bucket, id)
|
||||
}}
|
||||
return client.Where
|
||||
}
|
||||
|
||||
// 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.
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// Design 47 G §700, A-74: the router's record of an ask is read under the indexer's module's name from the state asks of the
|
||||
// seat operator-channel, in the bucket of the seat's holder on the seat list, and carries the ask's warrant.
|
||||
func TestTheRoutersRecordIsReadFromTheBucketOfTheSeatsHolder(t *testing.T) {
|
||||
var asked []string
|
||||
call := func(key string, body any) (json.RawMessage, error) {
|
||||
asked = append(asked, key)
|
||||
return json.RawMessage(`{"seats":[{"seat":"node-packet-filter","scope":"node","node":"n1","holders":[{"module":"pf","node":"n1"}]},` +
|
||||
`{"seat":"operator-channel","scope":"mesh","holders":[{"module":"ops-router","node":"n2"}]}],"holds":2}`), nil
|
||||
}
|
||||
var bucket, id string
|
||||
record := func(b, i string) (*stdio.StateEntry, error) {
|
||||
bucket, id = b, i
|
||||
return &stdio.StateEntry{Key: "mesh-delivery-indexer." + i, Value: json.RawMessage(`{"state":"chosen","warrant":{"ask":"a1","asker":"mesh-delivery-indexer","choice":"approve"}}`)}, nil
|
||||
}
|
||||
state, w, err := routerRecordOf("mesh-delivery-indexer", call, record)("a1")
|
||||
if err != nil || state != "chosen" || w == nil || w.Ask != "a1" {
|
||||
t.Fatalf("Where = %q, %+v, %v; want the state chosen and the warrant of a1", state, w, err)
|
||||
}
|
||||
if bucket != "ops-router_asks" || id != "a1" || len(asked) != 1 || asked[0] != "seat:mesh-module-assigner.assignments" {
|
||||
t.Errorf("read %q key %q after asking %v; want the holder's bucket ops-router_asks, the id, one question of the seat mesh-module-assigner", bucket, id, asked)
|
||||
}
|
||||
// No holder on the seat list: nothing is read, and the warrant stays unapplied with the reason said.
|
||||
none := func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"seats":[],"holds":0}`), nil }
|
||||
if _, _, err := routerRecordOf("mesh-delivery-indexer", none, record)("a1"); err == nil || !strings.Contains(err.Error(), "no module holds the seat operator-channel") {
|
||||
t.Errorf("with no holder Where answered %v; want the reason", err)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
NamedDrop(ctx context.Context, fingerprint, member, by, why string) index.TakenIn
|
||||
RegistryAct(ctx context.Context, a index.TakeInAsk) index.TakenIn
|
||||
}
|
||||
|
||||
@@ -54,6 +55,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
|
||||
} else if a.CandidateGroup != "" && a.Drop {
|
||||
asked, r = "drop of a candidate's catalog half", ix.DropCandidateGraph(ctx, a.CandidateGroup, a.CandidateHeads, a.By, a.Why)
|
||||
a.Module = r.Module
|
||||
} else if a.NamedFingerprint != "" && a.Drop {
|
||||
asked, r = "drop of a member of a fingerprint's names", ix.NamedDrop(ctx, a.NamedFingerprint, a.NamedMember, a.By, a.Why)
|
||||
a.Module = r.Module
|
||||
} else if a.Fingerprints {
|
||||
asked, r = "computing of fingerprints again", ix.FingerprintsAgain(ctx, a.Module)
|
||||
} else if a.RecoverTries {
|
||||
|
||||
@@ -5,6 +5,7 @@ go 1.26.0
|
||||
require (
|
||||
git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogkey v0.0.0-00010101000000-000000000000
|
||||
@@ -15,6 +16,7 @@ require (
|
||||
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
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/npmpackageregistry v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
)
|
||||
|
||||
// Design 47 G §668, §692: manifest-accepted's problems are the validator's strings, the same the verdict writes as
|
||||
// warnings, in the same order; a mended manifest says an empty list, never null.
|
||||
func TestManifestAcceptedProblemsAreTheVerdictsWarnings(t *testing.T) {
|
||||
warnings := []string{`requires "amqp", which no module provides`, "a second problem"}
|
||||
got := acceptedOf("sha256:aa", meshcatalog.VerdictsByDigestValue{Module: "m", Judge: "j", Warnings: warnings}, 3, []byte(`{"module":"m","version":"1.0"}`))
|
||||
if !reflect.DeepEqual(got.Problems, warnings) {
|
||||
t.Errorf("problems %v, want the verdict's warnings %v", got.Problems, warnings)
|
||||
}
|
||||
mended := acceptedOf("sha256:bb", meshcatalog.VerdictsByDigestValue{Module: "m", Judge: "j"}, 4, []byte(`{"module":"m","version":"2.0"}`))
|
||||
raw, _ := json.Marshal(mended)
|
||||
if !strings.Contains(string(raw), `"problems":[]`) {
|
||||
t.Errorf("a mended manifest's event is %s, want an empty problems list", raw)
|
||||
}
|
||||
}
|
||||
@@ -13,19 +13,21 @@ import (
|
||||
|
||||
// Each entry carries its artifacts' package names (design 47 G §676, §491, §478, §504, §412): at take-in the indexer reads
|
||||
// each artifact's package name from the revision's own files at the artifact's path, the module of its go.mod for a Go
|
||||
// package, the name of its package.json for an npm package, the manifest's name of an image artifact, and writes it into
|
||||
// package, the name of its package.json for an npm package, the module's and the artifact's names joined by / for an image artifact (§694), and writes it into
|
||||
// the entry's artifact as package, {ecosystem, name}. The manifest validator judges §491 from these fields.
|
||||
|
||||
// Package ecosystems of an artifact's package (§676).
|
||||
const (
|
||||
EcosystemGo = "go"
|
||||
EcosystemNpm = "npm"
|
||||
// EcosystemImage is an image artifact's package ecosystem (§694).
|
||||
EcosystemImage = "image"
|
||||
)
|
||||
|
||||
// ArtifactRefOf is the artifact reference of a module's artifact, <module>:<artifact>.
|
||||
func ArtifactRefOf(module, artifact string) string { return module + ":" + artifact }
|
||||
|
||||
// withArtifacts gives e its manifest's artifacts, by reference, with the package name of each package artifact (an image's name is an open design question: none is written). A
|
||||
// withArtifacts gives e its manifest's artifacts, by reference, with the package name of each package artifact (an image's is its module's name and its artifact's, §694). A
|
||||
// name the forge does not answer is said and left out of the artifact, for the next entry to read again (the cache keeps only
|
||||
// what was read); a manifest that does not decode leaves the entry as it is.
|
||||
func (ix *Indexer) withArtifacts(ctx context.Context, e meshcatalog.CatalogEntry, m catalogue.Manifest) meshcatalog.CatalogEntry {
|
||||
@@ -45,10 +47,13 @@ func (ix *Indexer) withArtifacts(ctx context.Context, e meshcatalog.CatalogEntry
|
||||
|
||||
// packageOf is the package name of artifact a of the entry e: read through the forge at the entry's commit, kept per
|
||||
// repository, commit, path and artifact so a later entry of the same commit reads nothing; nil for an artifact that names
|
||||
// none (a program, an image, an archive) or whose name cannot be read now.
|
||||
// none (a program, an archive) or whose name cannot be read now.
|
||||
func (ix *Indexer) packageOf(ctx context.Context, e meshcatalog.CatalogEntry, a catalogue.Artifact) *meshcatalog.ArtifactPackage {
|
||||
switch a.Kind {
|
||||
case catalogue.ArtifactPackage:
|
||||
case catalogue.ArtifactImage:
|
||||
// An image's package name is its module's name and its artifact's joined by `/`, read from no file (§694).
|
||||
return &meshcatalog.ArtifactPackage{Ecosystem: EcosystemImage, Name: e.Module + "/" + a.Name}
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -102,6 +102,9 @@ type Indexer struct {
|
||||
Revisions Objects
|
||||
// EntryIndex is the state entry-index (entry_verb.go, design 47 G §675); nil keeps none.
|
||||
EntryIndex Store
|
||||
// NamedSets is the state named: which want or running build names a fingerprint (named.go, §675); nil keeps none, and no
|
||||
// entry is then kept for a name.
|
||||
NamedSets *Named
|
||||
// SayCatalogRevision says the seat mesh-catalog's event catalog-revision for a revision kept; nil says nothing.
|
||||
SayCatalogRevision func(ctx context.Context, e meshcatalog.CatalogRevisionEvent) error
|
||||
// SayFill says the seat mesh-catalog's event filling, full or room of the catalog graph revisions, under the duplicate
|
||||
|
||||
@@ -63,8 +63,26 @@ func TestAnEntryCarriesItsArtifactsPackageNames(t *testing.T) {
|
||||
}
|
||||
got[a.Ref] = name
|
||||
}
|
||||
want := map[string]string{"m:gopkg": "go git.example/m/go", "m:jspkg": "npm @example/m", "m:img": "none", "m:prog": "none"}
|
||||
want := map[string]string{"m:gopkg": "go git.example/m/go", "m:jspkg": "npm @example/m", "m:img": "image m/img", "m:prog": "none"}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("artifacts' packages %v, want %v (§676)", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §694: an image artifact's package is {image, <module>/<artifact>}, read from no file, so two modules
|
||||
// with an image artifact of one name carry two package names.
|
||||
func TestAnImageArtifactsPackageNameIsItsModulesAndItsOwn(t *testing.T) {
|
||||
ix := &Indexer{}
|
||||
names := map[string]bool{}
|
||||
for _, module := range []string{"a", "b"} {
|
||||
m := catalogue.Manifest{Module: module, Build: &catalogue.Build{Artifacts: []catalogue.Artifact{{Name: "server", Kind: catalogue.ArtifactImage}}}}
|
||||
e := ix.withArtifacts(context.Background(), meshcatalog.CatalogEntry{Module: module}, m)
|
||||
if len(e.Artifacts) != 1 || e.Artifacts[0].Package == nil || e.Artifacts[0].Package.Ecosystem != "image" || e.Artifacts[0].Package.Name != module+"/server" {
|
||||
t.Fatalf("module %s: artifacts %+v, want the package {image, %s/server} (§694)", module, e.Artifacts, module)
|
||||
}
|
||||
names[e.Artifacts[0].Package.Name] = true
|
||||
}
|
||||
if len(names) != 2 {
|
||||
t.Errorf("two modules' image artifacts of one name carry %v, want two package names", names)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -286,6 +286,9 @@ func (ix *Indexer) keepEntries(ctx context.Context, entries map[string]meshcatal
|
||||
ix.logf("catalog graph: revision %d of source revision %d is not kept: %v", rev.Revision, src, err)
|
||||
return err
|
||||
}
|
||||
if _, err := ix.removeUnlisted(ctx); err != nil {
|
||||
ix.logf("catalog graph: the entries nothing lists or names any more are not removed: %v", err) // at the first revision after none does (§675)
|
||||
}
|
||||
if after, capacity, err := ix.Revisions.Fill(ctx); fillErr == nil && err == nil && capacity > 0 && used*5 < capacity*4 && after*5 >= capacity*4 {
|
||||
ix.sayFill(ctx, meshcatalog.EventFilling, fmt.Sprintf("%d", rev.Revision), meshcatalog.Fill{Bytes: after, Cap: capacity})
|
||||
}
|
||||
|
||||
@@ -104,16 +104,18 @@ func EntryTool(o Objects, index Store) stdio.Tool {
|
||||
// EntriesAnswer is what indexer_entries answers: for a fingerprint its key and its object's digest, and whether the object
|
||||
// is kept; with none given, every fingerprint the state holds.
|
||||
type EntriesAnswer struct {
|
||||
Fingerprint string `json:"fingerprint,omitempty"`
|
||||
Digest string `json:"digest,omitempty"`
|
||||
Kept bool `json:"kept,omitempty"`
|
||||
Keys []string `json:"keys,omitempty"`
|
||||
Fingerprint string `json:"fingerprint,omitempty"`
|
||||
Digest string `json:"digest,omitempty"`
|
||||
Kept bool `json:"kept,omitempty"`
|
||||
// Named is the set of `<node>.<module>` whose want or running build names the fingerprint (named.go, §675).
|
||||
Named []string `json:"named,omitempty"`
|
||||
Keys []string `json:"keys,omitempty"`
|
||||
}
|
||||
|
||||
// EntriesTool is indexer_entries (class read): one fingerprint's key and the digest of its object, or every key of
|
||||
// entry-index with none given.
|
||||
func EntriesTool(o Objects, index Store) stdio.Tool {
|
||||
return stdio.Tool{Name: EntryIndexTool, Description: "Read the indexer's entry-index: the digest of the catalog graph entry of a fingerprint and whether its object is kept; with no fingerprint, every key it holds.",
|
||||
func EntriesTool(o Objects, index, named Store) stdio.Tool {
|
||||
return stdio.Tool{Name: EntryIndexTool, Description: "Read the indexer's entry-index: the digest of the catalog graph entry of a fingerprint, whether its object is kept and the set of `<node>.<module>` whose want or running build names it (the state named, which keeps the entry while it is not empty); with no fingerprint, every key it holds. indexer_named_drop removes one member.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{"fingerprint": map[string]any{"type": "string", "description": "the entry's fingerprint; none lists every key"}}},
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
ctx := context.Background()
|
||||
@@ -138,6 +140,11 @@ func EntriesTool(o Objects, index Store) stdio.Tool {
|
||||
out.Kept = true
|
||||
}
|
||||
}
|
||||
if named != nil {
|
||||
if out.Named, err = NamedMembers(ctx, named, fp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}}
|
||||
}
|
||||
|
||||
@@ -20,7 +20,9 @@ func refusedOf(digest string, v meshcatalog.VerdictsByDigestValue) meshcatalog.M
|
||||
// acceptedEvent is the body of manifest-accepted for the verdict v accepting the manifest m of digest, its bus-state
|
||||
// entries whole (novox/hq ADR 0595 §4); an empty list when m asks for none.
|
||||
func acceptedEvent(digest string, v meshcatalog.VerdictsByDigestValue, m catalogue.Manifest) meshcatalog.ManifestAcceptedEvent {
|
||||
return meshcatalog.ManifestAcceptedEvent{V: keys.Version, Module: v.Module, Digest: digest, Judge: v.Judge, BusState: busStateOf(m.BusState)}
|
||||
// problems are the manifest validator's strings, the verdict's warnings in their order, an empty list when none (design 47 G §668, §692).
|
||||
problems := append([]string{}, v.Warnings...)
|
||||
return meshcatalog.ManifestAcceptedEvent{V: keys.Version, Module: v.Module, Digest: digest, Judge: v.Judge, BusState: busStateOf(m.BusState), Problems: problems}
|
||||
}
|
||||
|
||||
// busStateOf is the manifest's bus-state entries in the event's form (the same JSON).
|
||||
|
||||
@@ -0,0 +1,399 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
)
|
||||
|
||||
// Kept while named (design 47 G §675, §693, §695): the indexer keeps, in its old state section named, one key per
|
||||
// fingerprint holding the set of `<node>.<module>` whose want (the seat mesh-wants' wanted) or running build (the node-engine
|
||||
// seat's running) names it, written by compare-and-set and capped at 4 MiB (max-mib, §695). An entry object is removed only
|
||||
// when no kept revision lists it and its set is empty. Both events are streams kept last per subject, read whole at the
|
||||
// start and then followed (§556), so the sets are the same whenever the indexer started.
|
||||
const (
|
||||
// NamedState is the old state section of the sets of members by fingerprint.
|
||||
NamedState = "named"
|
||||
// NamedBucket is the bucket of the state.
|
||||
NamedBucket = Module + "_" + NamedState
|
||||
// NamedDropTool is the tool that removes one member from one fingerprint's set by hand (class act).
|
||||
NamedDropTool = "indexer_named_drop"
|
||||
// NamedWanted and NamedRunning name the two streams the sets are read from, for Named.ReadDone.
|
||||
NamedWanted = "wanted"
|
||||
NamedRunning = "running"
|
||||
)
|
||||
|
||||
// NamedValue is one key of named: the fingerprint and the members that name it, sorted.
|
||||
type NamedValue struct {
|
||||
V int `json:"v"`
|
||||
Fp string `json:"fp"`
|
||||
Members []string `json:"members"`
|
||||
}
|
||||
|
||||
// NamedKey is the key of a fingerprint in named, the fingerprint escaped as a key's name is (§73).
|
||||
func NamedKey(fingerprint string) string { return keys.EscapeName(fingerprint) }
|
||||
|
||||
// NamedMember is the member of a node's module in a set.
|
||||
func NamedMember(node, module string) string { return node + "." + module }
|
||||
|
||||
// namedAt is the fingerprint one source last named for a member, and when it said so.
|
||||
type namedAt struct {
|
||||
fp string
|
||||
at time.Time
|
||||
}
|
||||
|
||||
// Named keeps the sets. What each member's want and running last named is held in memory only, read again at every start from
|
||||
// the two streams, so a member whose fingerprint moved while the indexer was away leaves the old one's set at the end of the
|
||||
// start's read (Reconcile).
|
||||
type Named struct {
|
||||
// Store is the state named.
|
||||
Store Store
|
||||
// Changed is called, with no lock held, after a set lost a member: an entry object it kept may be removed now (§675); nil
|
||||
// calls nothing.
|
||||
Changed func(ctx context.Context)
|
||||
Logf func(format string, args ...any)
|
||||
|
||||
mu sync.Mutex
|
||||
want map[string]namedAt
|
||||
run map[string]namedAt
|
||||
gone map[string]time.Time // the want of a member removed (§693), when
|
||||
reads map[string]bool // the streams read whole at the start
|
||||
ready bool
|
||||
}
|
||||
|
||||
// NewNamed is the sets over store.
|
||||
func NewNamed(store Store, logf func(string, ...any)) *Named {
|
||||
if logf == nil {
|
||||
logf = func(string, ...any) {}
|
||||
}
|
||||
return &Named{Store: store, Logf: logf, want: map[string]namedAt{}, run: map[string]namedAt{}, gone: map[string]time.Time{}, reads: map[string]bool{}}
|
||||
}
|
||||
|
||||
// Ready says the two streams were read whole and the sets reconciled with them: until then no entry object is removed, so a
|
||||
// restart reads named whole before it removes anything (§675).
|
||||
func (n *Named) Ready() bool {
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
return n.ready
|
||||
}
|
||||
|
||||
// HearWanted applies one wanted event, said on mesh.seat.mesh-wants.event.wanted.<node>.<module>: the want names its fingerprint,
|
||||
// or, with removed, names none and leaves the member out of every set whatever its node last said running (§693).
|
||||
func (n *Named) HearWanted(ctx context.Context, data []byte) error {
|
||||
var e meshwants.WantedEvent
|
||||
if err := json.Unmarshal(data, &e); err != nil {
|
||||
return fmt.Errorf("a wanted that does not read: %w", err)
|
||||
}
|
||||
if e.Node == "" || e.Module == "" || e.Fp == "" {
|
||||
return errors.New("a wanted names no node, module or fingerprint")
|
||||
}
|
||||
var at time.Time
|
||||
if e.At != nil {
|
||||
at = *e.At
|
||||
}
|
||||
return n.apply(ctx, NamedMember(e.Node, e.Module), true, e.Fp, at, e.Removed)
|
||||
}
|
||||
|
||||
// HearRunning applies one running event: the build it names; a stopped one, or one with no fingerprint, names none (design 47
|
||||
// G §365).
|
||||
func (n *Named) HearRunning(ctx context.Context, data []byte) error {
|
||||
var r nodeengine.Running
|
||||
if err := json.Unmarshal(data, &r); err != nil {
|
||||
return fmt.Errorf("a running that does not read: %w", err)
|
||||
}
|
||||
if r.Node == "" || r.Module == "" {
|
||||
return errors.New("a running names no node or module")
|
||||
}
|
||||
fp := r.Fingerprint
|
||||
if r.Stopped {
|
||||
fp = ""
|
||||
}
|
||||
return n.apply(ctx, NamedMember(r.Node, r.Module), false, fp, r.At, false)
|
||||
}
|
||||
|
||||
// apply changes what one source names for the member and writes the sets it moves; one at a time, so the two sources never
|
||||
// write one set from a stale view of each other.
|
||||
func (n *Named) apply(ctx context.Context, member string, wanted bool, fp string, at time.Time, removed bool) error {
|
||||
n.mu.Lock()
|
||||
moved, err := n.applyLocked(ctx, member, wanted, fp, at, removed)
|
||||
n.mu.Unlock()
|
||||
if moved && n.Changed != nil {
|
||||
n.Changed(ctx)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (n *Named) applyLocked(ctx context.Context, member string, wanted bool, fp string, at time.Time, removed bool) (bool, error) {
|
||||
src := n.run
|
||||
if wanted {
|
||||
src = n.want
|
||||
}
|
||||
old, had := src[member]
|
||||
if had && at.Before(old.at) {
|
||||
return false, nil // an event older than the one heard changes nothing
|
||||
}
|
||||
var touched []string
|
||||
if had {
|
||||
touched = append(touched, old.fp)
|
||||
}
|
||||
switch {
|
||||
case removed:
|
||||
// The node left or the module was unplaced: the want goes, and with it what the node last said running.
|
||||
if r, ok := n.run[member]; ok {
|
||||
touched = append(touched, r.fp)
|
||||
delete(n.run, member)
|
||||
}
|
||||
delete(n.want, member)
|
||||
n.gone[member] = at
|
||||
touched = append(touched, fp)
|
||||
case !n.gone[member].IsZero() && !at.After(n.gone[member]):
|
||||
return false, nil // an event older than the removal of the want it belongs to
|
||||
case fp == "":
|
||||
delete(src, member)
|
||||
default:
|
||||
if wanted {
|
||||
delete(n.gone, member)
|
||||
}
|
||||
src[member] = namedAt{fp: fp, at: at}
|
||||
touched = append(touched, fp)
|
||||
}
|
||||
return n.writeSets(ctx, member, touched)
|
||||
}
|
||||
|
||||
// holds says whether either source names fp for the member.
|
||||
func (n *Named) holds(member, fp string) bool {
|
||||
return n.want[member].fp == fp || n.run[member].fp == fp
|
||||
}
|
||||
|
||||
// writeSets makes each fingerprint's set hold the member exactly when a source names it; it answers whether a set lost one.
|
||||
func (n *Named) writeSets(ctx context.Context, member string, fps []string) (lost bool, err error) {
|
||||
slices.Sort(fps)
|
||||
fps = slices.Compact(fps)
|
||||
var errs []error
|
||||
for _, fp := range fps {
|
||||
if fp == "" {
|
||||
continue
|
||||
}
|
||||
in := n.holds(member, fp)
|
||||
changed := false
|
||||
err := write(ctx, n.Store, NamedKey(fp), func(old []byte, found bool) ([]byte, bool, error) {
|
||||
var v NamedValue
|
||||
if found {
|
||||
if err := json.Unmarshal(old, &v); err != nil {
|
||||
return nil, false, fmt.Errorf("%s.%s does not read: %w", NamedState, NamedKey(fp), err)
|
||||
}
|
||||
}
|
||||
has := slices.Contains(v.Members, member)
|
||||
changed = has != in
|
||||
switch {
|
||||
case in && !has:
|
||||
v.Members = append(v.Members, member)
|
||||
case !in && has:
|
||||
v.Members = slices.DeleteFunc(v.Members, func(m string) bool { return m == member })
|
||||
default:
|
||||
return nil, false, nil
|
||||
}
|
||||
if len(v.Members) == 0 {
|
||||
return nil, true, nil
|
||||
}
|
||||
slices.Sort(v.Members)
|
||||
v.V, v.Fp = keys.Version, fp
|
||||
data, err := json.Marshal(v)
|
||||
return data, false, err
|
||||
})
|
||||
if err != nil {
|
||||
errs = append(errs, err)
|
||||
continue
|
||||
}
|
||||
if changed && !in {
|
||||
lost = true
|
||||
}
|
||||
}
|
||||
return lost, errors.Join(errs...)
|
||||
}
|
||||
|
||||
// ReadDone says the stream (NamedWanted or NamedRunning) was read whole at the start; with both read, every key is reconciled
|
||||
// with what the two named, a member whose build moved while the indexer was away leaving its old set, and the sets are ready.
|
||||
func (n *Named) ReadDone(ctx context.Context, stream string) error {
|
||||
n.mu.Lock()
|
||||
n.reads[stream] = true
|
||||
if n.ready || !n.reads[NamedWanted] || !n.reads[NamedRunning] {
|
||||
n.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
lost, err := n.reconcileLocked(ctx)
|
||||
if err == nil {
|
||||
n.ready = true
|
||||
}
|
||||
n.mu.Unlock()
|
||||
if (lost || err == nil) && n.Changed != nil {
|
||||
n.Changed(ctx) // the sets are ready: an entry kept back until now may go
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (n *Named) reconcileLocked(ctx context.Context) (bool, error) {
|
||||
all, err := n.Store.Keys(ctx)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("the state %s cannot be listed: %w", NamedState, err)
|
||||
}
|
||||
var errs []error
|
||||
lost := false
|
||||
for _, key := range all {
|
||||
err := write(ctx, n.Store, key, func(old []byte, found bool) ([]byte, bool, error) {
|
||||
var v NamedValue
|
||||
if !found || json.Unmarshal(old, &v) != nil {
|
||||
return nil, false, nil
|
||||
}
|
||||
kept := slices.DeleteFunc(slices.Clone(v.Members), func(m string) bool { return !n.holds(m, v.Fp) })
|
||||
if len(kept) == len(v.Members) {
|
||||
return nil, false, nil
|
||||
}
|
||||
lost = true
|
||||
if len(kept) == 0 {
|
||||
return nil, true, nil
|
||||
}
|
||||
v.Members = kept
|
||||
data, err := json.Marshal(v)
|
||||
return data, false, err
|
||||
})
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s.%s: %w", NamedState, key, err))
|
||||
}
|
||||
}
|
||||
return lost, errors.Join(errs...)
|
||||
}
|
||||
|
||||
// Members is the set of a fingerprint, sorted; none is an empty list.
|
||||
func (n *Named) Members(ctx context.Context, fingerprint string) ([]string, error) {
|
||||
return NamedMembers(ctx, n.Store, fingerprint)
|
||||
}
|
||||
|
||||
// NamedMembers reads a fingerprint's set from the state.
|
||||
func NamedMembers(ctx context.Context, st Store, fingerprint string) ([]string, error) {
|
||||
value, _, found, err := st.Get(ctx, NamedKey(fingerprint))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the state %s cannot be read: %w", NamedState, err)
|
||||
}
|
||||
if !found {
|
||||
return nil, nil
|
||||
}
|
||||
var v NamedValue
|
||||
if err := json.Unmarshal(value, &v); err != nil {
|
||||
return nil, fmt.Errorf("%s.%s does not read: %w", NamedState, NamedKey(fingerprint), err)
|
||||
}
|
||||
return v.Members, nil
|
||||
}
|
||||
|
||||
// IsNamed says whether a want or a running build names the fingerprint. An unreadable state names it, so no entry is removed on
|
||||
// a read that failed.
|
||||
func (n *Named) IsNamed(ctx context.Context, fingerprint string) (bool, error) {
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
m, err := n.Members(ctx, fingerprint)
|
||||
return err != nil || len(m) > 0, err
|
||||
}
|
||||
|
||||
// Drop removes one member from one fingerprint's set by hand (indexer_named_drop), for a member a want or a running build
|
||||
// named that the mesh no longer holds; the next event of that member is applied as any is. It answers whether the member was in
|
||||
// the set.
|
||||
func (n *Named) Drop(ctx context.Context, fingerprint, member string) (bool, error) {
|
||||
n.mu.Lock()
|
||||
dropped := false
|
||||
err := write(ctx, n.Store, NamedKey(fingerprint), func(old []byte, found bool) ([]byte, bool, error) {
|
||||
var v NamedValue
|
||||
if !found {
|
||||
return nil, false, nil
|
||||
}
|
||||
if err := json.Unmarshal(old, &v); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
if !slices.Contains(v.Members, member) {
|
||||
return nil, false, nil
|
||||
}
|
||||
dropped = true
|
||||
v.Members = slices.DeleteFunc(v.Members, func(m string) bool { return m == member })
|
||||
if len(v.Members) == 0 {
|
||||
return nil, true, nil
|
||||
}
|
||||
data, err := json.Marshal(v)
|
||||
return data, false, err
|
||||
})
|
||||
if dropped && err == nil {
|
||||
if n.want[member].fp == fingerprint {
|
||||
delete(n.want, member)
|
||||
}
|
||||
if n.run[member].fp == fingerprint {
|
||||
delete(n.run, member)
|
||||
}
|
||||
}
|
||||
n.mu.Unlock()
|
||||
if dropped && err == nil && n.Changed != nil {
|
||||
n.Changed(ctx)
|
||||
}
|
||||
return dropped, err
|
||||
}
|
||||
|
||||
// NamedDrop is indexer_named_drop answered: the member leaves the fingerprint's set, and an entry nothing names now may go.
|
||||
func (ix *Indexer) NamedDrop(ctx context.Context, fingerprint, member, by, why string) TakenIn {
|
||||
r := TakenIn{Module: "named " + fingerprint}
|
||||
if ix.NamedSets == nil {
|
||||
r.Answer = "the indexer keeps no state " + NamedState + ": nothing dropped"
|
||||
return r
|
||||
}
|
||||
dropped, err := ix.NamedSets.Drop(ctx, fingerprint, member)
|
||||
switch {
|
||||
case err != nil:
|
||||
r.Answer = member + " could not be dropped from " + fingerprint + ": " + err.Error()
|
||||
case !dropped:
|
||||
r.Answer = member + " is not in the set of " + fingerprint + ": nothing dropped"
|
||||
default:
|
||||
r.Stored = true
|
||||
r.Answer = fmt.Sprintf("%s is dropped from the set of %s by %s (%s)", member, fingerprint, by, why)
|
||||
}
|
||||
ix.logf("%s", r.Answer)
|
||||
return r
|
||||
}
|
||||
|
||||
// NamedTools is indexer_named_drop (class act): the tools bundle writes an ask in takeins and the indexer, the state's owner,
|
||||
// removes the member; record says the hand act in the hand-act log before it is asked.
|
||||
func NamedTools(ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(what, why string) error) []stdio.Tool {
|
||||
return []stdio.Tool{{Name: NamedDropTool,
|
||||
Description: "Remove one member (`<node>.<module>`) from the set of a fingerprint in the indexer's state named, which keeps the entry object of " +
|
||||
"the fingerprint while a want or a running build names it (design 47 G §675): for a member the mesh no longer holds, " +
|
||||
"which indexer_entries shows. A hand act, recorded through mesh-hand-acts before it is asked; an entry nothing lists or names " +
|
||||
"any more is removed.",
|
||||
Input: map[string]any{"type": "object", "required": []string{"fingerprint", "member", "why"}, "properties": map[string]any{
|
||||
"fingerprint": map[string]any{"type": "string", "description": "the entry's fingerprint"},
|
||||
"member": map[string]any{"type": "string", "description": "the member, <node>.<module>, as indexer_entries names it"},
|
||||
"why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}},
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
str := func(k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
|
||||
fp, member, why := str("fingerprint"), str("member"), str("why")
|
||||
if fp == "" || member == "" || why == "" {
|
||||
return nil, errors.New("name the fingerprint (fingerprint) and the member (member) and say why (why): a drop is a hand act, recorded through mesh-hand-acts")
|
||||
}
|
||||
data, err := json.Marshal(TakeInAsk{V: 1, NamedFingerprint: fp, NamedMember: member, Drop: true, Why: why, By: stdio.CallerOf(args), At: now().UTC()})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := record("dropped "+member+" from the names of the entry "+fp+" by hand", why); err != nil {
|
||||
return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was dropped: %w", err)
|
||||
}
|
||||
a, err := ask("named."+NamedKey(fp)+"."+keys.EscapeName(member), data)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a.Result, nil
|
||||
}}}
|
||||
}
|
||||
@@ -910,7 +910,7 @@ 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, NamedSets: 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...) }
|
||||
|
||||
@@ -1131,7 +1131,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", "Registry", "SeedModules"} {
|
||||
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules"} {
|
||||
named[f] = true
|
||||
}
|
||||
ty := reflect.TypeOf(Indexer{})
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
)
|
||||
|
||||
// The findings of the registrations (design 47 G §704, §8): the stream registration-history past 80 % full, and the seed
|
||||
// that waits for the operator.
|
||||
const (
|
||||
// HistoryFillingKind and HistoryFillingID are the finding of registration-history past HistoryFillingPercent (scope seat).
|
||||
HistoryFillingKind = "registration-history-filling"
|
||||
HistoryFillingID = "mesh-catalog.registration-history"
|
||||
// HistoryFillingPercent is the fill at which the finding is written.
|
||||
HistoryFillingPercent = 80
|
||||
// SeedWaitsKind and SeedWaitsID are the finding of a seed that waits for a warrant (scope seat), cleared by the warrant
|
||||
// and kept after a Decline or an expiry (§8).
|
||||
SeedWaitsKind = "seed-waits"
|
||||
SeedWaitsID = "mesh-catalog.seed"
|
||||
)
|
||||
|
||||
// HistoryFill writes the finding registration-history-filling when the stream holds used bytes of its max, 80 % or more,
|
||||
// and clears it under that; a stream with no cap says nothing (design 47 G §704).
|
||||
func (ix *Indexer) HistoryFill(ctx context.Context, used, max int64) {
|
||||
if ix.Findings == nil || max <= 0 {
|
||||
return
|
||||
}
|
||||
if used*100 >= max*HistoryFillingPercent {
|
||||
f := ix.observed(mcc.ScopeSeat, HistoryFillingID, HistoryFillingKind,
|
||||
fmt.Sprintf("registration-history holds %d of %d bytes", used, max),
|
||||
"The registration history is filling",
|
||||
fmt.Sprintf("The stream registration-history keeps every version of every registration and refuses a new one when full, so a registration change whose version cannot be kept fails. It holds %d of its %d bytes.", used, max),
|
||||
"indexer."+HistoryFillingKind)
|
||||
if err := ix.Findings.Write(ctx, f); err != nil {
|
||||
ix.logf("registrations: the finding %s is not written: %v", HistoryFillingKind, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err := ix.clearFinding(ctx, mcc.ScopeSeat, HistoryFillingID, HistoryFillingKind, "registration-history is under 80 % full"); err != nil {
|
||||
ix.logf("registrations: the finding %s is not cleared: %v", HistoryFillingKind, err)
|
||||
}
|
||||
}
|
||||
|
||||
// SeedWaits writes the finding seed-waits naming the ask when the first registrations wait, and clears it when the seed is
|
||||
// asked or written (design 47 G §704, §8).
|
||||
func (ix *Indexer) SeedWaits(ctx context.Context, ask string, waiting bool) {
|
||||
if ix.Findings == nil {
|
||||
return
|
||||
}
|
||||
if waiting {
|
||||
f := ix.observed(mcc.ScopeSeat, SeedWaitsID, SeedWaitsKind,
|
||||
"the first registrations wait on the operator (ask "+ask+")",
|
||||
"The first registrations wait for the operator",
|
||||
"The ask "+ask+" carries the first registrations and ended without a warrant; "+SeedAgainTool+" raises it again.",
|
||||
"indexer."+SeedWaitsKind)
|
||||
if err := ix.Findings.Write(ctx, f); err != nil {
|
||||
ix.logf("registrations: the finding %s is not written: %v", SeedWaitsKind, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err := ix.clearFinding(ctx, mcc.ScopeSeat, SeedWaitsID, SeedWaitsKind, "the first registrations are asked or written (ask "+ask+")"); err != nil {
|
||||
ix.logf("registrations: the finding %s is not cleared: %v", SeedWaitsKind, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
)
|
||||
|
||||
// Design 47 G §704: registration-history past 80 % full is the finding registration-history-filling (scope seat, id
|
||||
// mesh-catalog.registration-history), cleared under it; the seed that waits is seed-waits (scope seat, id mesh-catalog.seed),
|
||||
// cleared by the warrant only.
|
||||
func TestRegistrationFindings(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
w := newFindings()
|
||||
ix := &Indexer{Findings: w}
|
||||
held := func(kind string) bool {
|
||||
for _, f := range w.Held() {
|
||||
if f.Scope == mcc.ScopeSeat && f.Kind == kind {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
ix.HistoryFill(ctx, 79, 100)
|
||||
if held(HistoryFillingKind) {
|
||||
t.Fatal("79 % full wrote the finding")
|
||||
}
|
||||
ix.HistoryFill(ctx, 80, 100)
|
||||
if !held(HistoryFillingKind) {
|
||||
t.Fatal("80 % full did not write registration-history-filling")
|
||||
}
|
||||
if f := w.Held()[0]; f.ID != "mesh-catalog.registration-history" {
|
||||
t.Errorf("the finding's id is %q", f.ID)
|
||||
}
|
||||
ix.HistoryFill(ctx, 10, 100)
|
||||
if held(HistoryFillingKind) {
|
||||
t.Fatal("the finding stands under 80 %")
|
||||
}
|
||||
ix.SeedWaits(ctx, "ask-1", true)
|
||||
if !held(SeedWaitsKind) {
|
||||
t.Fatal("a seed that waits wrote no finding")
|
||||
}
|
||||
ix.SeedWaits(ctx, "ask-1", false)
|
||||
if held(SeedWaitsKind) {
|
||||
t.Fatal("the warrant did not clear seed-waits")
|
||||
}
|
||||
}
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
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"
|
||||
@@ -123,6 +124,9 @@ type Registry struct {
|
||||
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)
|
||||
// Placement reads a module's placements through the seat mesh-module-assigner's verb placement: one per watched branch a row
|
||||
// of its assignments names, with the nodes that run the module from it (design 47 G §704, A-62).
|
||||
Placement func(module string) ([]mma.ModulePlacement, 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
|
||||
@@ -205,7 +209,8 @@ 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).
|
||||
// forbiddenBranch refuses the two names the catalog's keys use (§343); the refusal's reason is `reserved branch name`
|
||||
// (design 47 G §704).
|
||||
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)
|
||||
@@ -218,24 +223,26 @@ 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) {
|
||||
// protect refuses a watched branch that is named as the catalog's keys are (reason `reserved branch name`, §704) or that
|
||||
// the forge does not protect (reason `unprotected branch`, §5), naming it; both empty when every branch is accepted.
|
||||
func (r *Registry) protect(repository string, branches []string) (reason, words string, err error) {
|
||||
for _, b := range branches {
|
||||
if w := forbiddenBranch(b); w != "" {
|
||||
return w, nil
|
||||
return meshcatalog.RegistrationAskedRefusedReservedBranchName, 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)
|
||||
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)
|
||||
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 meshcatalog.RegistrationAskedRefusedUnprotectedBranch,
|
||||
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
|
||||
return "", "", nil
|
||||
}
|
||||
|
||||
// stepLines names each step status a policy moves, one line each.
|
||||
@@ -414,6 +421,22 @@ func (r *Registry) Propose(ctx context.Context, verb, caller string, change json
|
||||
if verb != verbRegister && !found {
|
||||
return refused(module, meshcatalog.RegistrationAskedRefusedUnknownModule, module+" is not registered: register it first"), nil
|
||||
}
|
||||
// An unwatch of a branch a placement names is refused, naming the placement (§704, A-62).
|
||||
if verb == verbUnwatch {
|
||||
if r.Placement == nil {
|
||||
return meshcatalog.RegistrationAsked{}, errors.New("the placements of a module are not asked of the seat mesh-module-assigner, so no branch is unwatched")
|
||||
}
|
||||
placed, err := r.Placement(module)
|
||||
if err != nil {
|
||||
return meshcatalog.RegistrationAsked{}, fmt.Errorf("the placements of %s are not read: %w", module, err)
|
||||
}
|
||||
for _, p := range placed {
|
||||
if p.Branch == branch {
|
||||
return refused(module, meshcatalog.RegistrationAskedRefusedPlaced,
|
||||
fmt.Sprintf("the branch %s of %s is placed: %s run it from that branch", branch, module, strings.Join(p.Nodes, ", "))), nil
|
||||
}
|
||||
}
|
||||
}
|
||||
// register with no branch takes the repository's default branch (§342).
|
||||
if verb == verbRegister {
|
||||
var in meshcatalog.RegisterInput
|
||||
@@ -467,12 +490,12 @@ func (r *Registry) Propose(ctx context.Context, verb, caller string, change json
|
||||
branches = []string{branch}
|
||||
}
|
||||
if len(branches) > 0 {
|
||||
words, err := r.protect(next.Origin, branches)
|
||||
reason, words, err := r.protect(next.Origin, branches)
|
||||
if err != nil {
|
||||
return meshcatalog.RegistrationAsked{}, err
|
||||
}
|
||||
if words != "" {
|
||||
return refused(module, meshcatalog.RegistrationAskedRefusedUnprotectedBranch, words), nil
|
||||
if reason != "" {
|
||||
return refused(module, reason, words), nil
|
||||
}
|
||||
}
|
||||
if len(lines) == 0 {
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
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"
|
||||
@@ -26,6 +27,7 @@ type regRig struct {
|
||||
whereErr error
|
||||
chosen map[string]string // ask id -> option the router's record holds
|
||||
missing map[string]bool
|
||||
placed map[string][]mma.ModulePlacement // module -> its placements, as the seat mesh-module-assigner's placement answers
|
||||
}
|
||||
|
||||
func newRegRig() *regRig {
|
||||
@@ -45,8 +47,9 @@ func newRegRig() *regRig {
|
||||
}
|
||||
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 },
|
||||
Placement: func(module string) ([]mma.ModulePlacement, error) { return g.placed[module], 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
|
||||
@@ -101,15 +104,12 @@ func TestRegistrationsAreWrittenOnTheirWarrantOnly(t *testing.T) {
|
||||
}{
|
||||
{"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")},
|
||||
{"candidate", meshcatalog.RegistrationAskedRefusedReservedBranchName, reg("a", meshcatalog.StepStatuses{}, "candidate")},
|
||||
{"entry", meshcatalog.RegistrationAskedRefusedReservedBranchName, reg("a", meshcatalog.StepStatuses{}, "entry")},
|
||||
} {
|
||||
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 err != nil || got.Refused == "" || got.Ask != "" || got.Words == "" || got.Refused != c.why {
|
||||
t.Fatalf("%s: answer %+v, %v: want a refusal for %q naming why and no ask (design 47 G §704)", c.name, got, err, c.why)
|
||||
}
|
||||
}
|
||||
if len(g.asks) != 0 {
|
||||
@@ -179,6 +179,13 @@ func TestRegistrationsAreWrittenOnTheirWarrantOnly(t *testing.T) {
|
||||
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")
|
||||
// A branch a placement names is not unwatched: refused as placed, naming the placement, and no ask is raised (§704, A-62).
|
||||
g.placed = map[string][]mma.ModulePlacement{"a": {{Branch: "beta", Nodes: []string{"n1", "n2"}}}}
|
||||
if p, err := g.r.Propose(ctx, "unwatch", "s@n", mustJSON(meshcatalog.WatchInput{Module: "a", Branch: "beta"})); err != nil || p.Refused != meshcatalog.RegistrationAskedRefusedPlaced ||
|
||||
p.Ask != "" || !strings.Contains(p.Words, "beta") || !strings.Contains(p.Words, "n1, n2") {
|
||||
t.Fatalf("unwatch of a placed branch: %+v, %v; want refused placed naming the placement", p, err)
|
||||
}
|
||||
g.placed = nil
|
||||
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 {
|
||||
|
||||
@@ -132,9 +132,15 @@ func ListRevisions(ctx context.Context, o Objects, candidates bool) (RevisionsAn
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// removeUnlisted deletes every entry no kept revision lists any more: an entry is stored once and shared, so it goes
|
||||
// only with the last revision that lists it. It answers the entries removed.
|
||||
func removeUnlisted(ctx context.Context, o Objects) ([]string, error) {
|
||||
// removeUnlisted deletes every entry no kept revision lists any more and no want or running build names (design 47 G §675): an
|
||||
// entry is stored once and shared, so it goes only with the last revision that lists it, and its entry-index key with it. It
|
||||
// removes none while the sets of the names are not ready (a restart reads named whole before it removes anything), and keeps
|
||||
// an entry whose set it cannot read. It answers the entries removed. The caller holds keepMu.
|
||||
func (ix *Indexer) removeUnlisted(ctx context.Context) ([]string, error) {
|
||||
if ix.NamedSets != nil && !ix.NamedSets.Ready() {
|
||||
return nil, nil
|
||||
}
|
||||
o := ix.Revisions
|
||||
list, err := ListRevisions(ctx, o, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -145,14 +151,74 @@ func removeUnlisted(ctx context.Context, o Objects) ([]string, error) {
|
||||
}
|
||||
var removed []string
|
||||
for _, e := range list.Unlisted {
|
||||
fp, digest := ix.entryFingerprint(ctx, e)
|
||||
if ix.NamedSets != nil && fp != "" {
|
||||
if named, err := ix.NamedSets.IsNamed(ctx, fp); named {
|
||||
if err != nil {
|
||||
ix.logf("catalog graph: %s is kept, the state %s cannot be read: %v", e, NamedState, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
}
|
||||
if err := o.Delete(ctx, e); err != nil {
|
||||
return removed, fmt.Errorf("%s: %w", e, err)
|
||||
}
|
||||
removed = append(removed, e)
|
||||
ix.forgetEntryIndex(ctx, fp, digest)
|
||||
}
|
||||
return removed, nil
|
||||
}
|
||||
|
||||
// entryFingerprint is the build-input fingerprint of an entry object and its digest; none when the object does not read.
|
||||
func (ix *Indexer) entryFingerprint(ctx context.Context, object string) (fp, digest string) {
|
||||
digest = strings.TrimPrefix(object, "entry.")
|
||||
data, err := ix.Revisions.Get(ctx, object)
|
||||
if err != nil {
|
||||
return "", digest
|
||||
}
|
||||
var e meshcatalog.CatalogEntry
|
||||
if json.Unmarshal(data, &e) != nil {
|
||||
return "", digest
|
||||
}
|
||||
return e.Fp, digest
|
||||
}
|
||||
|
||||
// forgetEntryIndex deletes the entry-index key of a fingerprint when it still holds the digest of the object just removed (§675).
|
||||
func (ix *Indexer) forgetEntryIndex(ctx context.Context, fp, digest string) {
|
||||
if ix.EntryIndex == nil || fp == "" {
|
||||
return
|
||||
}
|
||||
err := write(ctx, ix.EntryIndex, EntryIndexKey(fp), func(old []byte, found bool) ([]byte, bool, error) {
|
||||
var v EntryIndexValue
|
||||
if !found || json.Unmarshal(old, &v) != nil || v.Digest != digest {
|
||||
return nil, false, nil
|
||||
}
|
||||
return nil, true, nil
|
||||
})
|
||||
if err != nil {
|
||||
ix.logf("catalog graph: the key of %s in %s is not removed: %v", fp, EntryIndexState, err)
|
||||
}
|
||||
}
|
||||
|
||||
// RemoveUnlisted removes the entries nothing names now (§675): called when a set of the names lost a member.
|
||||
func (ix *Indexer) RemoveUnlisted(ctx context.Context) {
|
||||
if ix.Revisions == nil {
|
||||
return
|
||||
}
|
||||
ix.keepMu.Lock()
|
||||
removed, err := ix.removeUnlisted(ctx)
|
||||
ix.keepMu.Unlock()
|
||||
if err != nil {
|
||||
ix.logf("catalog graph: the entries nothing names are not removed: %v", err)
|
||||
}
|
||||
if len(removed) > 0 {
|
||||
ix.logf("catalog graph: %d entries removed, no revision lists them and no want or running build names them", len(removed))
|
||||
if _, err := ix.RoomAgain(ctx); err != nil {
|
||||
ix.logf("catalog graph: room is not evaluated: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// DropCandidateGraph drops a candidate's catalog half by hand (indexer_drop_candidate_graph, design 47: for a candidate
|
||||
// whose gate event never came), the entries no other revision lists with it, and says room when that takes the store
|
||||
// below its filling mark, which takes in a merge held for objects full (RoomAgain).
|
||||
@@ -171,7 +237,7 @@ func (ix *Indexer) DropCandidateGraph(ctx context.Context, group, heads, by, why
|
||||
err := ix.Revisions.Delete(ctx, name)
|
||||
var removed []string
|
||||
if err == nil {
|
||||
removed, err = removeUnlisted(ctx, ix.Revisions)
|
||||
removed, err = ix.removeUnlisted(ctx)
|
||||
}
|
||||
ix.keepMu.Unlock()
|
||||
if err != nil {
|
||||
|
||||
@@ -34,8 +34,8 @@ func RevisionCallers(entries []meshcatalog.CatalogEntry, holder string) stdio.Ca
|
||||
// graph revision, the trunk's newest when the input names none, a kept trunk revision by its number, or a candidate's by
|
||||
// its group and heads, with one page of its entries whole, by module name. The caller is the runtime's proof: it is
|
||||
// answered only when the newest trunk revision's entry of its module declares the verb in invokes, and the holder is
|
||||
// refused too. With no trunk revision kept yet nobody is answered and the reader waits (ErrNoRevision), never an empty
|
||||
// catalog.
|
||||
// refused too. With no trunk revision kept yet the answer is `none-yet` alone (design 47 G §530), never an empty
|
||||
// catalog and never an error.
|
||||
func Revision(ctx context.Context, o Objects, holder, caller string, in meshcatalog.RevisionInput) (meshcatalog.RevisionAnswer, error) {
|
||||
var out meshcatalog.RevisionAnswer
|
||||
newest, ok, err := newestTrunk(ctx, o)
|
||||
@@ -43,7 +43,8 @@ func Revision(ctx context.Context, o Objects, holder, caller string, in meshcata
|
||||
return out, fmt.Errorf("the catalog graph revisions cannot be read: %w", err)
|
||||
}
|
||||
if !ok {
|
||||
return out, ErrNoRevision
|
||||
// An answer is told by its fields (design 47 G §530): no trunk revision kept yet is `none-yet`, no error.
|
||||
return meshcatalog.RevisionAnswer{NoneYet: true}, nil
|
||||
}
|
||||
_, trunkEntries, err := ReadRevision(ctx, o, RevisionObject(newest))
|
||||
if err != nil {
|
||||
|
||||
@@ -146,7 +146,8 @@ func (r *Registry) ProposeSeed(ctx context.Context, modules []SeedModule, again
|
||||
return "", err
|
||||
}
|
||||
if r.SeedWaits != nil {
|
||||
r.SeedWaits(p.Ask, false)
|
||||
// The seed waits on the operator while its ask is open; only the warrant clears the finding (design 47 G §704).
|
||||
r.SeedWaits(p.Ask, true)
|
||||
}
|
||||
return p.Ask, nil
|
||||
}
|
||||
|
||||
@@ -70,6 +70,10 @@ type TakeInAsk struct {
|
||||
CandidateGroup string `json:"candidate-group,omitempty"`
|
||||
CandidateHeads string `json:"candidate-heads,omitempty"`
|
||||
Drop bool `json:"drop,omitempty"`
|
||||
// NamedFingerprint and NamedMember, with Drop, ask one member to be removed from that fingerprint's set of the state named
|
||||
// (indexer_named_drop, named.go, design 47 G §675).
|
||||
NamedFingerprint string `json:"named-fingerprint,omitempty"`
|
||||
NamedMember string `json:"named-member,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"`
|
||||
@@ -91,7 +95,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 == "" && a.Registry == nil {
|
||||
if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.NamedFingerprint == "" && a.Registry == nil {
|
||||
return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try")
|
||||
}
|
||||
return a, nil
|
||||
|
||||
@@ -620,8 +620,8 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { //
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := Revision(context.Background(), verbObjects, Module, "n1.node-tools", meshcatalog.RevisionInput{}); !errors.Is(err, ErrNoRevision) {
|
||||
t.Errorf("revision with none kept answered %v, want it to wait", err)
|
||||
if got, err := Revision(context.Background(), verbObjects, Module, "n1.node-tools", meshcatalog.RevisionInput{}); err != nil || !got.NoneYet || got.Revision != nil || len(got.Entries) != 0 {
|
||||
t.Errorf("revision with none kept answered %+v, %v, want none-yet alone with no error (design 47 G §530)", got, err)
|
||||
}
|
||||
keep(meshcatalog.CatalogRevision{Revision: 3, Kind: "trunk"}, map[string]bool{"node-tools": true, Module: true})
|
||||
keep(meshcatalog.CatalogRevision{Revision: 4, Kind: "candidate", Group: "feat/x", Heads: "0123456789ab"}, map[string]bool{"node-tools": true, "mesh-grants": true})
|
||||
|
||||
@@ -0,0 +1,205 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// Design 47 G §675 (Kept while named), §693 (a removed want): the state named holds, per fingerprint, the set of
|
||||
// `<node>.<module>` whose want or running build names it.
|
||||
|
||||
var namedBase = time.Date(2026, 11, 11, 20, 0, 0, 0, time.UTC)
|
||||
|
||||
func namedFp(c string) string { return "src2:" + strings.Repeat(c, 64) }
|
||||
|
||||
type namedEvent struct {
|
||||
wanted bool
|
||||
node string
|
||||
module string
|
||||
fp string
|
||||
min int
|
||||
removed bool
|
||||
stopped bool
|
||||
}
|
||||
|
||||
func (e namedEvent) apply(t *testing.T, n *Named) {
|
||||
t.Helper()
|
||||
at := namedBase.Add(time.Duration(e.min) * time.Minute)
|
||||
var data []byte
|
||||
var err error
|
||||
if e.wanted {
|
||||
data, err = json.Marshal(meshwants.WantedEvent{V: 1, Node: e.node, Module: e.module, Fp: e.fp, Revision: 1, Removed: e.removed, At: &at})
|
||||
if err == nil {
|
||||
err = n.HearWanted(context.Background(), data)
|
||||
}
|
||||
} else {
|
||||
data, err = json.Marshal(map[string]any{"v": 1, "node": e.node, "module": e.module, "fingerprint": e.fp, "stopped": e.stopped, "at": at})
|
||||
if err == nil {
|
||||
err = n.HearRunning(context.Background(), data)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("%+v: %v", e, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNamedHoldsTheMembersWhoseWantOrRunningBuildNamesAFingerprint(t *testing.T) {
|
||||
a, b := namedFp("a"), namedFp("b")
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
events []namedEvent
|
||||
want map[string][]string
|
||||
}{
|
||||
{"a want and a running build of one member name one fingerprint once", []namedEvent{
|
||||
{wanted: true, node: "n1", module: "m", fp: a, min: 1}, {node: "n1", module: "m", fp: a, min: 2}}, map[string][]string{a: {"n1.m"}, b: nil}},
|
||||
{"two nodes of a module are two members", []namedEvent{
|
||||
{wanted: true, node: "n1", module: "m", fp: a, min: 1}, {wanted: true, node: "n2", module: "m", fp: a, min: 1}}, map[string][]string{a: {"n1.m", "n2.m"}}},
|
||||
{"a want that moves on leaves the old fingerprint's set", []namedEvent{
|
||||
{wanted: true, node: "n1", module: "m", fp: a, min: 1}, {wanted: true, node: "n1", module: "m", fp: b, min: 2}}, map[string][]string{a: nil, b: {"n1.m"}}},
|
||||
{"a running build still naming the old one keeps it there", []namedEvent{
|
||||
{wanted: true, node: "n1", module: "m", fp: a, min: 1}, {node: "n1", module: "m", fp: a, min: 1},
|
||||
{wanted: true, node: "n1", module: "m", fp: b, min: 2}}, map[string][]string{a: {"n1.m"}, b: {"n1.m"}}},
|
||||
{"a stopped running names nothing", []namedEvent{
|
||||
{node: "n1", module: "m", fp: a, min: 1}, {node: "n1", module: "m", stopped: true, min: 2}}, map[string][]string{a: nil}},
|
||||
{"a want removed empties the member whatever its node last said running (§693)", []namedEvent{
|
||||
{wanted: true, node: "n1", module: "m", fp: a, min: 1}, {node: "n1", module: "m", fp: a, min: 1},
|
||||
{wanted: true, node: "n1", module: "m", fp: a, min: 3, removed: true}}, map[string][]string{a: nil}},
|
||||
{"a running older than the removal does not bring the member back", []namedEvent{
|
||||
{wanted: true, node: "n1", module: "m", fp: a, min: 1, removed: true}, {node: "n1", module: "m", fp: a, min: 0}}, map[string][]string{a: nil}},
|
||||
{"an event older than the one heard changes nothing", []namedEvent{
|
||||
{wanted: true, node: "n1", module: "m", fp: b, min: 5}, {wanted: true, node: "n1", module: "m", fp: a, min: 4}}, map[string][]string{a: nil, b: {"n1.m"}}},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
st := newMem()
|
||||
n := NewNamed(st, nil)
|
||||
for _, e := range c.events {
|
||||
e.apply(t, n)
|
||||
}
|
||||
for fp, want := range c.want {
|
||||
got, err := n.Members(context.Background(), fp)
|
||||
if err != nil || !slices.Equal(got, want) {
|
||||
t.Errorf("the set of %.9s = %v, %v; want %v", fp, got, err, want)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// §675: an entry no kept revision lists is removed only when its set is empty, and not before the start's read of both streams
|
||||
// ended; its entry-index key goes with it.
|
||||
func TestAnEntryNoRevisionListsStaysWhileAWantNamesItAndGoesAfterTheWantMovesOn(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fp := namedFp("c")
|
||||
n := NewNamed(newMem(), nil)
|
||||
ix := &Indexer{Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), NamedSets: n, Now: func() time.Time { return namedBase }}
|
||||
n.Changed = ix.RemoveUnlisted
|
||||
if err := ix.keepEntries(ctx, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c", Fp: fp, Manifest: map[string]any{}}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ix.Revisions.Delete(ctx, "trunk.1"); err != nil { // the revision that listed it is removed
|
||||
t.Fatal(err)
|
||||
}
|
||||
found := func() bool {
|
||||
_, err := Entry(ctx, ix.Revisions, ix.EntryIndex, meshcatalog.EntryInput{Fingerprint: fp})
|
||||
return err == nil
|
||||
}
|
||||
|
||||
ix.RemoveUnlisted(ctx)
|
||||
if !found() {
|
||||
t.Fatal("an entry was removed before the start's read of the wants and the running builds ended")
|
||||
}
|
||||
namedEvent{wanted: true, node: "n1", module: "avahi", fp: fp, min: 1}.apply(t, n)
|
||||
if err := n.ReadDone(ctx, NamedWanted); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !found() {
|
||||
t.Fatal("an entry was removed with one stream not read")
|
||||
}
|
||||
if err := n.ReadDone(ctx, NamedRunning); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !found() {
|
||||
t.Fatal("an entry a want names was removed")
|
||||
}
|
||||
namedEvent{wanted: true, node: "n1", module: "avahi", fp: namedFp("d"), min: 2}.apply(t, n)
|
||||
if found() {
|
||||
t.Error("the entry stays after the want moved on")
|
||||
}
|
||||
if keys, _ := ix.EntryIndex.Keys(ctx); len(keys) != 0 {
|
||||
t.Errorf("entry-index still holds %v for the removed entry", keys)
|
||||
}
|
||||
}
|
||||
|
||||
// §675: a restart reads named whole before it removes anything, and a member whose build moved while the indexer was away
|
||||
// leaves its old set when both streams are read.
|
||||
func TestARestartReconcilesTheSetsWithWhatTheStreamsHoldNow(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
st := newMem()
|
||||
old, now := namedFp("e"), namedFp("f")
|
||||
before := NewNamed(st, nil)
|
||||
namedEvent{wanted: true, node: "n1", module: "m", fp: old, min: 1}.apply(t, before)
|
||||
namedEvent{wanted: true, node: "n2", module: "m", fp: old, min: 1}.apply(t, before)
|
||||
|
||||
after := NewNamed(st, nil) // the next start: the state kept, nothing in memory
|
||||
namedEvent{wanted: true, node: "n1", module: "m", fp: now, min: 2}.apply(t, after)
|
||||
if got, _ := after.Members(ctx, old); !slices.Equal(got, []string{"n1.m", "n2.m"}) {
|
||||
t.Fatalf("a restart removed from %v before it read both streams", got)
|
||||
}
|
||||
namedEvent{wanted: true, node: "n2", module: "m", fp: old, min: 1}.apply(t, after)
|
||||
_ = after.ReadDone(ctx, NamedWanted)
|
||||
_ = after.ReadDone(ctx, NamedRunning)
|
||||
if got, _ := after.Members(ctx, old); !slices.Equal(got, []string{"n2.m"}) {
|
||||
t.Errorf("the old set holds %v after the read, want only the member still naming it", got)
|
||||
}
|
||||
if !after.Ready() {
|
||||
t.Error("the sets are not ready after both streams were read")
|
||||
}
|
||||
}
|
||||
|
||||
// §675: indexer_named_drop removes one member by hand, says the hand act first, and indexer_entries reads the set.
|
||||
func TestNamedDropRemovesAMemberByHandAndEntriesReadsTheSet(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fp := namedFp("9")
|
||||
st := newMem()
|
||||
n := NewNamed(st, nil)
|
||||
namedEvent{wanted: true, node: "n1", module: "m", fp: fp, min: 1}.apply(t, n)
|
||||
namedEvent{wanted: true, node: "n2", module: "m", fp: fp, min: 1}.apply(t, n)
|
||||
ix := &Indexer{NamedSets: n, Logf: func(string, ...any) {}}
|
||||
var acts []string
|
||||
asked := func(key string, value []byte) (TakeInAsk, error) {
|
||||
a, err := DecodeTakeIn(value)
|
||||
if err != nil {
|
||||
return a, err
|
||||
}
|
||||
r := ix.NamedDrop(ctx, a.NamedFingerprint, a.NamedMember, a.By, a.Why)
|
||||
a.Result = &r
|
||||
return a, nil
|
||||
}
|
||||
tool := NamedTools(asked, func() time.Time { return namedBase }, func(what, why string) error { acts = append(acts, what+" | "+why); return nil })[0]
|
||||
if _, err := tool.Run(map[string]any{"fingerprint": fp, "member": "n1.m"}); err == nil {
|
||||
t.Error("a drop with no why was asked")
|
||||
}
|
||||
if _, err := tool.Run(map[string]any{"fingerprint": fp, "member": "n1.m", "why": "its node is gone"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(acts) != 1 || !strings.Contains(acts[0], "n1.m") {
|
||||
t.Errorf("the hand acts = %v, want the drop of n1.m said once", acts)
|
||||
}
|
||||
if got, _ := n.Members(ctx, fp); !slices.Equal(got, []string{"n2.m"}) {
|
||||
t.Errorf("the set holds %v, want n2.m alone", got)
|
||||
}
|
||||
read, err := EntriesTool(newMemObjects(1<<10), newMem(), st).Run(map[string]any{"fingerprint": fp})
|
||||
if err != nil || fmt.Sprint(read.(EntriesAnswer).Named) != "[n2.m]" {
|
||||
t.Errorf("indexer_entries = %+v, %v, want the named set [n2.m]", read, err)
|
||||
}
|
||||
var _ stdio.Tool = tool
|
||||
}
|
||||
@@ -2929,7 +2929,7 @@ func caseAStateNameIsLocalAndOneToken(t *testing.T) {
|
||||
{`{"module":"a.b","version":"1.0","state":["s"]}`, `no dot`},
|
||||
{`{"module":"a","version":"1.0","reads":["bindings"]}`, `a read is <module>.<name>`},
|
||||
{`{"module":"a","version":"1.0","reads":["a.s"]}`, `which is its own state`},
|
||||
{`{"module":"a","version":"1.0","state":[{"name":"s","shared":true}]}`, `{name, history, ttl-seconds, per-machine, key-ttl, read-only, form, catch-up}`},
|
||||
{`{"module":"a","version":"1.0","state":[{"name":"s","shared":true}]}`, `{name, history, ttl-seconds, per-machine, key-ttl, read-only, max-mib, form, catch-up}`},
|
||||
} {
|
||||
_, err := ParseManifest([]byte(c.manifest))
|
||||
if c.says == "" {
|
||||
|
||||
@@ -30,6 +30,7 @@ type state struct {
|
||||
Name string `json:"name"`
|
||||
History int `json:"history,omitempty"`
|
||||
KeyTTL bool `json:"key-ttl,omitempty"`
|
||||
MaxMiB int `json:"max-mib,omitempty"`
|
||||
}
|
||||
|
||||
type artifact struct {
|
||||
@@ -74,6 +75,7 @@ type manifest struct {
|
||||
State []state `json:"state"`
|
||||
Reads []string `json:"reads"`
|
||||
Invokes []string `json:"invokes"`
|
||||
Uses []string `json:"uses"`
|
||||
OwnSecrets map[string]string `json:"own-secrets"`
|
||||
SecretsOwner string `json:"secrets-owner"`
|
||||
Resources []map[string]any `json:"resources"`
|
||||
@@ -118,12 +120,17 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if own := slices.DeleteFunc(slices.Clone(m.Emits), func(e string) bool { return strings.HasPrefix(e, views.EventPrefix) }); !reflect.DeepEqual(own, []string{takein.TakeInPendingEvent, takein.TakeInTakenEvent}) {
|
||||
t.Errorf("emits = %v, want the two take-in events (the verdict events are the seat mesh-catalog's)", m.Emits)
|
||||
}
|
||||
// Design 47 G §700, A-74: the router's record of an ask is read through the seat operator-channel, which the indexer uses,
|
||||
// and its holder is asked of the seat mesh-module-assigner.
|
||||
if !slices.Equal(m.Uses, []string{"operator-channel"}) || !slices.Contains(m.Invokes, "seat:mesh-module-assigner.assignments") || !slices.Contains(m.Invokes, "seat:mesh-module-assigner.placement") {
|
||||
t.Errorf("uses = %v, invokes = %v, want the seat operator-channel used and seat:mesh-module-assigner.assignments invoked", m.Uses, m.Invokes)
|
||||
}
|
||||
// 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard.
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "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: index.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},
|
||||
{Name: index.RegistrationsState, History: 1}, {Name: index.ProposalsState, 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, MaxMiB: 8}, {Name: index.NamedState, History: 1, MaxMiB: 4}, // design 47 G §695
|
||||
{Name: index.RegistrationsState, History: 1}, {Name: index.ProposalsState, History: 1, MaxMiB: 4}}; !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,
|
||||
@@ -159,7 +166,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-conditions.findings-asked", "mesh-delivery-assigner.views.assignments",
|
||||
"mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**",
|
||||
"mesh-delivery.source-moved", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
"mesh-delivery.source-moved", "mesh-wants.wanted", "node-engine.running", "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" {
|
||||
@@ -171,7 +178,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", "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) {
|
||||
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-module-assigner.assignments", "seat:mesh-module-assigner.placement", "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
|
||||
@@ -193,9 +200,9 @@ 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
|
||||
index.EntryIndexTool, index.NamedDropTool, // 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") {
|
||||
if !reflect.DeepEqual(m.Tools, want) || len(want) != 46 || 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.
|
||||
@@ -311,8 +318,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if !slices.Contains(m.Emits, takein.TakeInPendingEvent) || !slices.Contains(m.Emits, takein.TakeInTakenEvent) {
|
||||
t.Errorf("emits = %v: want both take-in events emitted", m.Emits)
|
||||
}
|
||||
// No bus, no other seat (ADR 0437 §4).
|
||||
for _, k := range []string{"logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
|
||||
// No bus, no other seat (ADR 0437 §4); it uses the seat operator-channel alone (design 47 G §700, A-74, above).
|
||||
for _, k := range []string{"logs", "provides", "requires", "defines-seats", "seat", "seats", "bus"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the indexer has none", k)
|
||||
}
|
||||
|
||||
@@ -66,6 +66,7 @@
|
||||
"indexer_recompute",
|
||||
"indexer_drop_candidate_graph",
|
||||
"indexer_entries",
|
||||
"indexer_named_drop",
|
||||
"indexer_proposals",
|
||||
"indexer_proposal_drop",
|
||||
"indexer_seed_again"
|
||||
@@ -82,6 +83,8 @@
|
||||
"mesh-delivery-checker.verdict",
|
||||
"mesh-delivery-forge-bridge.forge.**",
|
||||
"mesh-delivery.source-moved",
|
||||
"mesh-wants.wanted",
|
||||
"node-engine.running",
|
||||
"operator-channel.decided",
|
||||
"npm-package-registry.published"
|
||||
],
|
||||
@@ -105,9 +108,10 @@
|
||||
{ "name": "seat-definitions", "history": 5 },
|
||||
{ "name": "pending", "history": 1 },
|
||||
{ "name": "unjudged", "history": 1 },
|
||||
{ "name": "entry-index", "history": 1 },
|
||||
{ "name": "entry-index", "history": 1, "max-mib": 8 },
|
||||
{ "name": "named", "history": 1, "max-mib": 4 },
|
||||
{ "name": "registrations", "history": 1 },
|
||||
{ "name": "proposals", "history": 1 }
|
||||
{ "name": "proposals", "history": 1, "max-mib": 4 }
|
||||
],
|
||||
"streams": [
|
||||
{ "name": "registration-history", "events": ["registration.*"], "retention": "history", "max-mib": 64 },
|
||||
@@ -148,9 +152,14 @@
|
||||
"mesh-delivery-forge-bridge.bridge_merged_message",
|
||||
"seat:node-engine.composition",
|
||||
"seat:mesh-delivery.source",
|
||||
"seat:mesh-module-assigner.assignments",
|
||||
"seat:mesh-module-assigner.placement",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-watchdogs.presence"
|
||||
],
|
||||
"uses": [
|
||||
"operator-channel"
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user