From 0f62358fb066a855f23ba18249c4f6ba25b0267e Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:49:57 +0100 Subject: [PATCH] indexer: the findings registration-history-filling past 80 % and seed-waits kept until the warrant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §704 (registration-history, seed findings), §8 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-indexer/main.go | 2 +- .../cmd/mesh-delivery-indexer/registry.go | 13 +++- .../internal/index/registration_findings.go | 66 +++++++++++++++++++ .../index/registration_findings_test.go | 48 ++++++++++++++ .../internal/index/seed.go | 3 +- 5 files changed, 128 insertions(+), 4 deletions(-) create mode 100644 modules/mesh-delivery-indexer/internal/index/registration_findings.go create mode 100644 modules/mesh-delivery-indexer/internal/index/registration_findings_test.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 54eb0e52..5beaf5bc 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -450,7 +450,7 @@ 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) + routerRecordUnread, ix, logger.Printf) ix.Registry = registry ix.SeedModules = func(context.Context) ([]index.SeedModule, error) { return nil, errors.New("the first registrations are not composed by this build: what the build seats answer steps with is not read yet (design question)") diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go index 176b0c6d..a16b9621 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go @@ -48,7 +48,7 @@ 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 { +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 +56,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) { @@ -107,8 +115,9 @@ 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} } diff --git a/modules/mesh-delivery-indexer/internal/index/registration_findings.go b/modules/mesh-delivery-indexer/internal/index/registration_findings.go new file mode 100644 index 00000000..ffe323a5 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/registration_findings.go @@ -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) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/registration_findings_test.go b/modules/mesh-delivery-indexer/internal/index/registration_findings_test.go new file mode 100644 index 00000000..be737f0a --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/registration_findings_test.go @@ -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") + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/seed.go b/modules/mesh-delivery-indexer/internal/index/seed.go index d25a4218..1ef74b8c 100644 --- a/modules/mesh-delivery-indexer/internal/index/seed.go +++ b/modules/mesh-delivery-indexer/internal/index/seed.go @@ -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 }