IX17, IX18: the indexer keeps the records pinning each catalog graph revision and removes a revision only on an event
State pins (4 MiB; one key per revision <branch>.<revision>, made, pins-stated) applied from a durable consumer of mesh-delivery's pinned, freed, heads-replaced and pins-stated, acknowledged after the apply; a revision is removed when no record pins it and it is not current, only while the consumer has nothing pending and after a pins-stated later than made; finding pin-missing; tools indexer_pins, indexer_unpin, indexer_consumer (the state consumers, written by mesh-sdk's process.ConsumerLedger). Builds: design 47 G §153, §178, §189, §215, §216, §579, A-169, A-171, A-176 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -18,6 +18,7 @@ import (
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/busstate"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
@@ -299,6 +300,11 @@ func main() {
|
||||
_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.NamedDropTool, Verb: "named-drop"})
|
||||
return err
|
||||
})...)
|
||||
tools = append(tools, index.PinsTools(keptRead{stdio.State(index.PinsState)}, revisions, asks, time.Now, func(what, why string) error {
|
||||
_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.UnpinTool, Verb: "unpin"})
|
||||
return err
|
||||
})...)
|
||||
tools = append(tools, index.ConsumerTools(keptRead{stdio.State(process.ConsumersState)})...)
|
||||
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
|
||||
|
||||
@@ -608,6 +608,11 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, bucket := range []string{index.PinsBucket, index.Module + "_" + process.ConsumersState} {
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.NamedBucket, History: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -1406,7 +1411,7 @@ func serveAssignerAssignments(t *testing.T, ctx context.Context, js jetstream.Je
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy,
|
||||
FilterSubjects: []string{"mesh.mod." + assignerHolderModule + ".event.views.>"}}); err != nil {
|
||||
FilterSubjects: []string{"mesh.mod." + assignerHolderModule + ".event.views.>", "mesh.seat.mesh-delivery.event.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
em := views.Emitter{Owner: assignerHolderModule, Subject: process.Account{Node: "n1", Module: assignerHolderModule}.EventSubject,
|
||||
@@ -1491,6 +1496,46 @@ func TestIndexerKeepsWhatAWantAndARunningBuildNameAndLetsAGoneOneLeave(t *testin
|
||||
}
|
||||
}
|
||||
|
||||
// TestIndexerAppliesThePinsEventsThroughItsDurableConsumer (design 47 G §153, §216, §579, A-169, A-171, A-176): a pinned of a revision the
|
||||
// indexer does not hold and a pins-stated, said on the seat mesh-delivery, are applied to the state pins through its durable consumer (the
|
||||
// key `<branch>.<revision>` marked missing, the record listed, `made` and `pins-stated` kept), and the consumer's facts reach the state
|
||||
// consumers for indexer_consumer.
|
||||
func TestIndexerAppliesThePinsEventsThroughItsDurableConsumer(t *testing.T) {
|
||||
js, ctx, _, _ := startIndexer(t, func(context.Context, buildLog) {})
|
||||
say := func(subject string, body any) {
|
||||
raw, _ := json.Marshal(body)
|
||||
if _, err := js.Publish(ctx, subject, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
say(meshdelivery.SubjectPinned, meshdelivery.Pins{Record: "r1", Milestone: "47/G", Catalog: []meshdelivery.PinsCatalogItem{{Branch: "main", Revision: 9}}})
|
||||
say(meshdelivery.SubjectPinsStated, meshdelivery.PinsStatedEvent{At: time.Now().UTC().Add(time.Minute)})
|
||||
pins, err := js.KeyValue(ctx, index.PinsBucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
consumers, err := js.KeyValue(ctx, index.Module+"_"+process.ConsumersState)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var v index.PinValue
|
||||
var f process.ConsumerFacts
|
||||
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
e, err := pins.Get(ctx, "main.9")
|
||||
c, cerr := consumers.Get(ctx, "n1_mesh-delivery-indexer")
|
||||
if _, serr := pins.Get(ctx, index.PinsStatedKey); err == nil && cerr == nil && serr == nil &&
|
||||
json.Unmarshal(e.Value(), &v) == nil && json.Unmarshal(c.Value(), &f) == nil && f.LastEvent == meshdelivery.SubjectPinsStated {
|
||||
break
|
||||
}
|
||||
}
|
||||
if !slices.Equal(v.Records, []string{"r1"}) || !v.Missing {
|
||||
t.Errorf("pins main.9 holds %+v within 20 s, want record r1 and missing", v)
|
||||
}
|
||||
if f.Stream != "EVENTS" || f.Acknowledged == 0 || f.LastEvent != meshdelivery.SubjectPinsStated {
|
||||
t.Errorf("consumers holds %+v, want the pins-stated applied on EVENTS", f)
|
||||
}
|
||||
}
|
||||
|
||||
// TestIndexerTakesInAWatchedBranchOtherThanTheFirstIntoItsOwnRevision: the seat's verb source lists a second watched branch,
|
||||
// beta, with the merge of mesh-catalog it names as its head (design 47 G §13, §14, §291, A-93); the indexer started with the
|
||||
// branch listed reads source.beta, takes the head's merged message in as a pure computation, and keeps branch.beta.1, which
|
||||
|
||||
@@ -439,6 +439,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
var fillFollow atomic.Pointer[versionsFill]
|
||||
// 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)
|
||||
pinsBucket := process.AwaitBucket(ctx, js, conn, index.PinsBucket, "no record's pin on a catalog graph revision is kept, so no revision is removed", logger.Printf)
|
||||
pinsStore := awaitedStore{b: pinsBucket}
|
||||
consumers := process.AwaitBucket(ctx, js, conn, index.Module+"_"+process.ConsumersState, "indexer_consumer reads no consumer", 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,
|
||||
@@ -451,8 +454,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
SayCatalogRevision: afterRevision(refreshingBuildModules(sayCatalogRevision(conn, cred), &buildFollow), &fillFollow),
|
||||
SayFill: sayFill(conn, cred),
|
||||
// The catalog's released versions (design 47 G §349): the state versions, bound when it appears, and the event version-released.
|
||||
Versions: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.VersionsBucket, "no released version is kept, so no package-published makes a version", logger.Printf)},
|
||||
BuildEdges: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.BuildEdgesBucket, "no build of another watched branch keeps its edges", logger.Printf)},
|
||||
Versions: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.VersionsBucket, "no released version is kept, so no package-published makes a version", logger.Printf)},
|
||||
BuildEdges: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.BuildEdgesBucket, "no build of another watched branch keeps its edges", logger.Printf)},
|
||||
Pins: pinsStore, PinsPending: pendingOf(js, cred),
|
||||
Inputs: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.InputsBucket, "no input of a run ask is kept, so the verb input answers none", logger.Printf)},
|
||||
SayVersionReleased: sayVersionReleased(conn, cred), SayVersionsFilled: sayVersionsFilled(conn, cred),
|
||||
// The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear;
|
||||
@@ -682,6 +686,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err := hub.Handle(meshdelivery.SubjectSourceMovedAll, func(m jetstream.Msg) { source.heard(ctx, m) }); err != nil {
|
||||
return err
|
||||
}
|
||||
pinsLedger := process.NewConsumerLedger(process.EventsStream, cred.EventConsumerName(), consumerSink(consumers), nil, logger.Printf)
|
||||
for _, subject := range pinsEvents {
|
||||
if err := hub.Handle(subject, func(m jetstream.Msg) { heardPins(ctx, ix, pinsLedger, m, logger.Printf) }); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Once the state is bound the pins are made and every kept revision judged (§579: at its start).
|
||||
pinsBucket.Ready(func(jetstream.KeyValue) {
|
||||
go func() {
|
||||
if err := ix.PinsMade(ctx); err != nil {
|
||||
logger.Printf("pins: the state %s is not made: %v", index.PinsState, err)
|
||||
return
|
||||
}
|
||||
ix.JudgePins(ctx)
|
||||
}()
|
||||
})
|
||||
versions := func(m jetstream.Msg) {
|
||||
if !heardVersion(ctx, ix, m, logger.Printf) {
|
||||
_ = m.Ack() // nothing else the indexer consumes rides the shared stream
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// The indexer's durable consumer of mesh-delivery's pinned, freed, heads-replaced and pins-stated (design 47 G §153, §216, §579):
|
||||
// each event is applied to the state pins, and only then acknowledged, so one said while the indexer was down is applied when it
|
||||
// comes back. The consumer's facts are kept in the state consumers (A-176) for indexer_consumer.
|
||||
|
||||
// pinsEvents are the subjects the indexer applies to its pins.
|
||||
var pinsEvents = []string{meshdelivery.SubjectPinned, meshdelivery.SubjectFreed, meshdelivery.SubjectHeadsReplaced, meshdelivery.SubjectPinsStated}
|
||||
|
||||
// pendingOf says how many messages of the module's durable consumer of the events stream have not been given to it yet.
|
||||
func pendingOf(js jetstream.JetStream, cred process.Account) func(context.Context) (uint64, error) {
|
||||
return func(ctx context.Context) (uint64, error) {
|
||||
cons, err := cred.EventConsumer(ctx, js, process.EventsStream)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
info, err := cons.Info(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return info.NumPending, nil
|
||||
}
|
||||
}
|
||||
|
||||
// consumerSink writes one key of the state consumers once its bucket is bound.
|
||||
func consumerSink(b *process.AwaitedBucket) process.ConsumerSink {
|
||||
return func(ctx context.Context, key string, value []byte) error {
|
||||
kv := b.KV()
|
||||
if kv == nil {
|
||||
return process.ErrNotYet
|
||||
}
|
||||
_, err := kv.Put(ctx, key, value)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// heardPins applies one pins event, acknowledges it, and judges the revisions kept: an event not applied is left unacknowledged, so
|
||||
// the bus gives it again.
|
||||
func heardPins(ctx context.Context, ix *index.Indexer, ledger *process.ConsumerLedger, m jetstream.Msg, logf func(string, ...any)) {
|
||||
ledger.Delivered(m)
|
||||
if err := ix.HeardPins(ctx, m.Subject(), m.Data()); err != nil {
|
||||
logf("pins: %s is not applied, left for the bus to give again: %v", m.Subject(), err)
|
||||
return
|
||||
}
|
||||
_ = m.Ack()
|
||||
var body struct {
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Data(), &body)
|
||||
ledger.Applied(m, body.At)
|
||||
ix.JudgePins(ctx)
|
||||
}
|
||||
@@ -28,6 +28,7 @@ type takerIn interface {
|
||||
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
|
||||
Unpin(ctx context.Context, revision, record, by, why string) index.TakenIn
|
||||
RegistryAct(ctx context.Context, a index.TakeInAsk) index.TakenIn
|
||||
}
|
||||
|
||||
@@ -58,6 +59,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
|
||||
} 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.UnpinRevision != "" {
|
||||
asked, r = "unpin of a record from a catalog graph revision", ix.Unpin(ctx, a.UnpinRevision, a.UnpinRecord, 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 {
|
||||
|
||||
@@ -181,11 +181,15 @@ type Indexer struct {
|
||||
saidAt map[string]string // the head of a repository whose failed read was said, by repository: said once per head
|
||||
failedAt map[string]string // the head of a repository whose tree lacks a directory asked, by repository: not asked again until the head changes
|
||||
// OwnBuild answers the fingerprint of the build this program runs, from the node-engine's seat; nil knows none.
|
||||
OwnBuild func(ctx context.Context) (string, error)
|
||||
underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go)
|
||||
judgeMu sync.Mutex // one judge pass at a time (verdicts.go)
|
||||
ownMu sync.Mutex // guards ownFP
|
||||
keepMu sync.Mutex // one catalog graph revision kept at a time (keepEntries)
|
||||
OwnBuild func(ctx context.Context) (string, error)
|
||||
underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go)
|
||||
judgeMu sync.Mutex // one judge pass at a time (verdicts.go)
|
||||
ownMu sync.Mutex // guards ownFP
|
||||
keepMu sync.Mutex // one catalog graph revision kept at a time (keepEntries)
|
||||
// Pins is the state pins (pins.go); nil keeps none. PinsPending says how many messages the indexer's durable consumer of the
|
||||
// events stream has not been given: a revision is removed only while none is pending (§579).
|
||||
Pins Store
|
||||
PinsPending func(ctx context.Context) (uint64, error)
|
||||
namesMu sync.Mutex // packageNames
|
||||
packageNames map[string]string // the package names read through the forge, by repository, commit, path and artifact (artifacts.go)
|
||||
catchingUp atomic.Bool // the first start's catch-up runs: its merges keep no revision until it ends (CatchUp)
|
||||
|
||||
@@ -348,6 +348,12 @@ func (ix *Indexer) keepBranchEntries(ctx context.Context, branch string, src int
|
||||
ix.logf("catalog graph: revision %d of source revision %d is not kept: %v", rev.Revision, src, err)
|
||||
return err
|
||||
}
|
||||
// The revision before this one in its branch ceases to be current (§579): it goes unless a record pins it.
|
||||
if removed, err := ix.judgePinsLocked(ctx); err != nil {
|
||||
ix.logf("pins: the revisions no record pins are not removed: %v", err)
|
||||
} else if len(removed) > 0 {
|
||||
ix.logf("pins: %d revisions removed (%s), no record pins them and none is current", len(removed), strings.Join(removed, ", "))
|
||||
}
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -910,7 +910,7 @@ func (ix *Indexer) dry() *dryRun {
|
||||
Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil,
|
||||
OwnBuildState: nil, Heard: nil, Placement: nil, RunningWhole: 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, NamedSets: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil, Inputs: nil,
|
||||
Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, NamedSets: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil, Inputs: nil, Pins: nil, PinsPending: 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", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "BuildEdges"} {
|
||||
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "BuildEdges", "Pins", "PinsPending"} {
|
||||
named[f] = true
|
||||
}
|
||||
ty := reflect.TypeOf(Indexer{})
|
||||
|
||||
@@ -0,0 +1,576 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// The indexer keeps the records pinning each catalog graph revision and removes a revision only on an event (design 47 G §153,
|
||||
// §178, §189, §215, §216, §579, A-169, A-171): its state pins holds one key per revision, `<branch>.<revision>` (the watched branch
|
||||
// and the revision's number, the branch escaped as a name is), whose value lists the delivery record ids that pin it, and two more
|
||||
// keys, `made` (written once, create-only, when the state is first used) and `pins-stated` (the `at` of the newest applied event of
|
||||
// that name). Pins count only once a pins-stated later than made is applied.
|
||||
const (
|
||||
// PinsState and PinsBucket are the state's name and its bucket; PinsCapMiB is its cap in the manifest (about 1,800 keys of about 0.2 KiB).
|
||||
PinsState = "pins"
|
||||
PinsBucket = Module + "_" + PinsState
|
||||
PinsHistory = 1
|
||||
PinsCapMiB = 4
|
||||
// PinsMadeKey and PinsStatedKey are the two keys of the state that are no revision's.
|
||||
PinsMadeKey = "made"
|
||||
PinsStatedKey = "pins-stated"
|
||||
// PinMissingKind is the finding of a pinned revision the indexer no longer holds.
|
||||
PinMissingKind = "pin-missing"
|
||||
// PinsTool reads the pins, UnpinTool removes one record's pin by hand.
|
||||
PinsTool = "indexer_pins"
|
||||
UnpinTool = "indexer_unpin"
|
||||
)
|
||||
|
||||
// PinValue is one revision's key: the records pinning it, and whether the revision was not held when a record pinned it.
|
||||
type PinValue struct {
|
||||
Records []string `json:"records"`
|
||||
Missing bool `json:"missing,omitempty"`
|
||||
}
|
||||
|
||||
// PinStamp is the value of the keys made and pins-stated.
|
||||
type PinStamp struct {
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// PinKey is a revision's key in the state pins.
|
||||
func PinKey(branch string, revision int64) string {
|
||||
return keys.EscapeName(branch) + "." + strconv.FormatInt(revision, 10)
|
||||
}
|
||||
|
||||
// ParsePinKey is the branch and the revision a key of the state pins names; false for made, pins-stated and what is no revision's key.
|
||||
func ParsePinKey(key string) (branch string, revision int64, ok bool) {
|
||||
i := strings.LastIndex(key, ".")
|
||||
if i <= 0 {
|
||||
return "", 0, false
|
||||
}
|
||||
n, err := strconv.ParseInt(key[i+1:], 10, 64)
|
||||
if err != nil {
|
||||
return "", 0, false
|
||||
}
|
||||
b, err := keys.UnescapeName(key[:i])
|
||||
if err != nil {
|
||||
return "", 0, false
|
||||
}
|
||||
return b, n, true
|
||||
}
|
||||
|
||||
// pinObject is the revision object a key names.
|
||||
func pinObject(key string) (string, bool) {
|
||||
branch, revision, ok := ParsePinKey(key)
|
||||
if !ok {
|
||||
return "", false
|
||||
}
|
||||
return RevisionObject(meshcatalog.CatalogRevision{Branch: branch, Revision: revision}), true
|
||||
}
|
||||
|
||||
func (ix *Indexer) pinsStamp(ctx context.Context, key string) (PinStamp, bool, error) {
|
||||
var s PinStamp
|
||||
v, _, found, err := ix.Pins.Get(ctx, key)
|
||||
if err != nil || !found {
|
||||
return s, false, err
|
||||
}
|
||||
if err := json.Unmarshal(v, &s); err != nil {
|
||||
return s, false, fmt.Errorf("%s.%s: %w", PinsState, key, err)
|
||||
}
|
||||
return s, true, nil
|
||||
}
|
||||
|
||||
// PinsMade writes the key made once, create-only, when the state is first used; a made already written stays.
|
||||
func (ix *Indexer) PinsMade(ctx context.Context) error {
|
||||
if ix.Pins == nil {
|
||||
return nil
|
||||
}
|
||||
data, err := json.Marshal(PinStamp{At: ix.now().UTC()})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := ix.Pins.Create(ctx, PinsMadeKey, data); err != nil && !errors.Is(err, ErrConflict) {
|
||||
return fmt.Errorf("%s.%s: %w", PinsState, PinsMadeKey, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// pinsCounted says the pins count: a pins-stated later than made is applied.
|
||||
func (ix *Indexer) pinsCounted(ctx context.Context) (bool, error) {
|
||||
made, ok, err := ix.pinsStamp(ctx, PinsMadeKey)
|
||||
if err != nil || !ok {
|
||||
return false, err
|
||||
}
|
||||
stated, ok, err := ix.pinsStamp(ctx, PinsStatedKey)
|
||||
if err != nil || !ok {
|
||||
return false, err
|
||||
}
|
||||
return stated.At.After(made.At), nil
|
||||
}
|
||||
|
||||
// HeardPins applies one event of the seat mesh-delivery the indexer's durable consumer gave it: pinned, freed, heads-replaced or
|
||||
// pins-stated, by its subject. An error leaves the event unacknowledged, so the bus gives it again; one applied twice changes nothing.
|
||||
func (ix *Indexer) HeardPins(ctx context.Context, subject string, body []byte) error {
|
||||
if ix.Pins == nil {
|
||||
return errors.New("the indexer keeps no state " + PinsState)
|
||||
}
|
||||
if err := ix.PinsMade(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
switch subject {
|
||||
case meshdelivery.SubjectPinned, meshdelivery.SubjectFreed:
|
||||
var p meshdelivery.Pins
|
||||
if err := json.Unmarshal(body, &p); err != nil {
|
||||
return fmt.Errorf("%s is not a pins body: %w", subject, err)
|
||||
}
|
||||
for _, c := range p.Catalog {
|
||||
key := PinKey(c.Branch, c.Revision)
|
||||
if subject == meshdelivery.SubjectPinned {
|
||||
if err := ix.pin(ctx, key, p.Record); err != nil {
|
||||
return err
|
||||
}
|
||||
} else if err := ix.unpin(ctx, key, p.Record); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
case meshdelivery.SubjectHeadsReplaced:
|
||||
var h meshdelivery.HeadsReplacedEvent
|
||||
if err := json.Unmarshal(body, &h); err != nil {
|
||||
return fmt.Errorf("%s is not a heads-replaced body: %w", subject, err)
|
||||
}
|
||||
ix.DropCandidateGraph(ctx, h.Group, h.Heads, "mesh-delivery", "its event heads-replaced: the candidate's heads were replaced or its pull request closed")
|
||||
case meshdelivery.SubjectPinsStated:
|
||||
var s meshdelivery.PinsStatedEvent
|
||||
if err := json.Unmarshal(body, &s); err != nil {
|
||||
return fmt.Errorf("%s is not a pins-stated body: %w", subject, err)
|
||||
}
|
||||
err := write(ctx, ix.Pins, PinsStatedKey, func(old []byte, found bool) ([]byte, bool, error) {
|
||||
var cur PinStamp
|
||||
if found && json.Unmarshal(old, &cur) == nil && !s.At.After(cur.At) {
|
||||
return nil, false, nil
|
||||
}
|
||||
data, err := json.Marshal(PinStamp{At: s.At.UTC()})
|
||||
return data, false, err
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("%s is no event of the pins", subject)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// pin adds a record's id to a revision's key by compare-and-set; a revision not held is marked missing and said.
|
||||
func (ix *Indexer) pin(ctx context.Context, key, record string) error {
|
||||
object, _ := pinObject(key)
|
||||
missing := false
|
||||
if ix.Revisions != nil {
|
||||
if _, err := ix.Revisions.Get(ctx, object); errors.Is(err, ErrNoObject) {
|
||||
missing = true
|
||||
} else if err != nil {
|
||||
return fmt.Errorf("%s cannot be read: %w", object, err)
|
||||
}
|
||||
}
|
||||
err := write(ctx, ix.Pins, key, func(old []byte, found bool) ([]byte, bool, error) {
|
||||
var v PinValue
|
||||
if found {
|
||||
if err := json.Unmarshal(old, &v); err != nil {
|
||||
return nil, false, fmt.Errorf("%s.%s: %w", PinsState, key, err)
|
||||
}
|
||||
}
|
||||
if !slices.Contains(v.Records, record) {
|
||||
v.Records = append(v.Records, record)
|
||||
slices.Sort(v.Records)
|
||||
}
|
||||
v.Missing = v.Missing || (missing && !found)
|
||||
data, err := json.Marshal(v)
|
||||
return data, false, err
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if missing {
|
||||
ix.sayPinMissing(ctx, key, record)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// unpin removes a record's id from a revision's key; a missing revision's key left with no record goes, and its finding.
|
||||
func (ix *Indexer) unpin(ctx context.Context, key, record string) error {
|
||||
var gone bool
|
||||
err := write(ctx, ix.Pins, key, func(old []byte, found bool) ([]byte, bool, error) {
|
||||
gone = false
|
||||
if !found {
|
||||
return nil, false, nil
|
||||
}
|
||||
var v PinValue
|
||||
if err := json.Unmarshal(old, &v); err != nil {
|
||||
return nil, false, fmt.Errorf("%s.%s: %w", PinsState, key, err)
|
||||
}
|
||||
v.Records = slices.DeleteFunc(v.Records, func(r string) bool { return r == record })
|
||||
if v.Missing && len(v.Records) == 0 {
|
||||
gone = true
|
||||
return nil, true, nil
|
||||
}
|
||||
data, err := json.Marshal(v)
|
||||
return data, false, err
|
||||
})
|
||||
if err == nil && gone {
|
||||
ix.clearPinMissing(ctx, key)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func pinFindingID(key string) string { return Module + ".pin." + key }
|
||||
|
||||
func (ix *Indexer) sayPinMissing(ctx context.Context, key, record string) {
|
||||
if ix.Findings == nil {
|
||||
return
|
||||
}
|
||||
object, _ := pinObject(key)
|
||||
f := ix.observed(mcc.ScopeModule, pinFindingID(key), PinMissingKind,
|
||||
fmt.Sprintf("the delivery record %s pins the catalog graph revision %s, which the indexer no longer holds", record, object),
|
||||
"A delivery record points at a catalog graph revision the indexer does not keep",
|
||||
"The indexer removed the revision, or never kept it, before the record pinned it. indexer_unpin removes the record's pin and clears this.", PinMissingKind)
|
||||
if err := ix.Findings.Write(ctx, f); err != nil {
|
||||
ix.logf("pins: the finding %s for %s is not written: %v", PinMissingKind, key, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (ix *Indexer) clearPinMissing(ctx context.Context, key string) {
|
||||
if ix.Findings == nil {
|
||||
return
|
||||
}
|
||||
if err := ix.clearFinding(ctx, mcc.ScopeModule, pinFindingID(key), PinMissingKind, "no record pins "+key+" any more"); err != nil {
|
||||
ix.logf("pins: the finding %s for %s is not cleared: %v", PinMissingKind, key, err)
|
||||
}
|
||||
}
|
||||
|
||||
// RevisionGroup says which watched branch's line a revision object belongs to, and its number; false for a candidate or an entry.
|
||||
func RevisionGroup(name string) (group string, revision int64, ok bool) {
|
||||
if n, ok := TrunkRevisionOf(name); ok {
|
||||
return "", n, true
|
||||
}
|
||||
if rest, ok := strings.CutPrefix(name, "branch."); ok {
|
||||
i := strings.LastIndex(rest, ".")
|
||||
if i > 0 {
|
||||
if n, err := strconv.ParseInt(rest[i+1:], 10, 64); err == nil {
|
||||
return rest[:i], n, true
|
||||
}
|
||||
}
|
||||
}
|
||||
return "", 0, false
|
||||
}
|
||||
|
||||
// CurrentRevisions are the kept revision objects that are current (§579): the newest of each watched branch, and every open
|
||||
// candidate's, with why each is.
|
||||
func CurrentRevisions(names []string) map[string]string {
|
||||
newest := map[string]int64{}
|
||||
for _, n := range names {
|
||||
if g, r, ok := RevisionGroup(n); ok && r > newest[g] {
|
||||
newest[g] = r
|
||||
}
|
||||
}
|
||||
out := map[string]string{}
|
||||
for _, n := range names {
|
||||
if g, r, ok := RevisionGroup(n); ok && r == newest[g] {
|
||||
out[n] = "the newest revision of its watched branch"
|
||||
} else if strings.HasPrefix(n, "candidate.") {
|
||||
out[n] = "an open candidate's"
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// JudgePins removes every kept revision no record pins and which is not current, entries no kept revision lists with them, and
|
||||
// says room; at the start, when the consumer is caught up and the pins count, it judges every kept revision (§579). It judges on
|
||||
// applying freed, on a revision ceasing to be current and on being caught up, never on a timer.
|
||||
func (ix *Indexer) JudgePins(ctx context.Context) {
|
||||
if ix.Pins == nil || ix.Revisions == nil {
|
||||
return
|
||||
}
|
||||
ix.keepMu.Lock()
|
||||
removed, err := ix.judgePinsLocked(ctx)
|
||||
var dropped []string
|
||||
if err == nil && len(removed) > 0 {
|
||||
dropped, err = ix.removeUnlisted(ctx)
|
||||
}
|
||||
ix.keepMu.Unlock()
|
||||
if err != nil {
|
||||
ix.logf("pins: the revisions no record pins are not removed: %v", err)
|
||||
}
|
||||
if len(removed) > 0 {
|
||||
ix.logf("pins: %d revisions removed (%s) with %d entries, no record pins them and none is current", len(removed), strings.Join(removed, ", "), len(dropped))
|
||||
if _, err := ix.RoomAgain(ctx); err != nil {
|
||||
ix.logf("pins: room is not evaluated: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// judgePinsLocked removes the revisions; the caller holds keepMu. It removes none while the consumer is not caught up or the pins do
|
||||
// not count yet.
|
||||
func (ix *Indexer) judgePinsLocked(ctx context.Context) ([]string, error) {
|
||||
if ix.PinsPending == nil {
|
||||
return nil, nil
|
||||
}
|
||||
pending, err := ix.PinsPending(ctx)
|
||||
if err != nil || pending != 0 {
|
||||
return nil, err
|
||||
}
|
||||
counted, err := ix.pinsCounted(ctx)
|
||||
if err != nil || !counted {
|
||||
return nil, err
|
||||
}
|
||||
names, err := ix.Revisions.Names(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
current := CurrentRevisions(names)
|
||||
pinned := map[string]bool{}
|
||||
pinKeys, err := ix.Pins.Keys(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, k := range pinKeys {
|
||||
object, ok := pinObject(k)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
v, _, found, err := ix.Pins.Get(ctx, k)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var pv PinValue
|
||||
if found && json.Unmarshal(v, &pv) == nil && len(pv.Records) > 0 {
|
||||
pinned[object] = true
|
||||
}
|
||||
}
|
||||
var removed []string
|
||||
for _, n := range names {
|
||||
if _, _, ok := RevisionGroup(n); !ok || pinned[n] {
|
||||
continue
|
||||
}
|
||||
if _, cur := current[n]; cur {
|
||||
continue
|
||||
}
|
||||
if err := ix.Revisions.Delete(ctx, n); err != nil {
|
||||
return removed, fmt.Errorf("%s: %w", n, err)
|
||||
}
|
||||
removed = append(removed, n)
|
||||
}
|
||||
for _, k := range pinKeys {
|
||||
object, ok := pinObject(k)
|
||||
if !ok || !slices.Contains(removed, object) {
|
||||
continue
|
||||
}
|
||||
if v, rev, found, err := ix.Pins.Get(ctx, k); err == nil && found {
|
||||
var pv PinValue
|
||||
if json.Unmarshal(v, &pv) == nil && len(pv.Records) == 0 {
|
||||
_ = ix.Pins.Delete(ctx, k, rev)
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.Strings(removed)
|
||||
return removed, nil
|
||||
}
|
||||
|
||||
// Unpin removes one record's pin from a revision by hand (indexer_unpin), for a freed that was lost, and the removal rule then applies.
|
||||
func (ix *Indexer) Unpin(ctx context.Context, revision, record, by, why string) TakenIn {
|
||||
r := TakenIn{Module: "pins " + revision}
|
||||
if ix.Pins == nil {
|
||||
r.Answer = "the indexer keeps no state " + PinsState + ": nothing unpinned"
|
||||
return r
|
||||
}
|
||||
if _, _, ok := ParsePinKey(revision); !ok {
|
||||
r.Answer = revision + " is no key of the state " + PinsState + " as indexer_pins names them: nothing unpinned"
|
||||
return r
|
||||
}
|
||||
if err := ix.unpin(ctx, revision, record); err != nil {
|
||||
r.Answer = "the pin of " + record + " on " + revision + " could not be removed: " + err.Error()
|
||||
ix.logf("%s", r.Answer)
|
||||
return r
|
||||
}
|
||||
r.Stored = true
|
||||
r.Answer = fmt.Sprintf("the pin of %s on %s is removed by %s (%s)", record, revision, by, why)
|
||||
ix.logf("%s", r.Answer)
|
||||
ix.JudgePins(ctx)
|
||||
return r
|
||||
}
|
||||
|
||||
// PinnedRevision is one revision of the state pins as indexer_pins lists it.
|
||||
type PinnedRevision struct {
|
||||
Revision string `json:"revision"`
|
||||
Object string `json:"object"`
|
||||
Records []string `json:"records"`
|
||||
Missing bool `json:"missing,omitempty"`
|
||||
Kept bool `json:"kept"`
|
||||
// Current says it is current, Why in what way.
|
||||
Current bool `json:"current,omitempty"`
|
||||
Why string `json:"why,omitempty"`
|
||||
}
|
||||
|
||||
// PinsAnswer is what indexer_pins answers.
|
||||
type PinsAnswer struct {
|
||||
Pins []PinnedRevision `json:"pins"`
|
||||
// Unpinned are the kept revisions no record pins, current or not, and Current why each current one is.
|
||||
Unpinned []string `json:"unpinned,omitempty"`
|
||||
Made *time.Time `json:"made,omitempty"`
|
||||
Stated *time.Time `json:"pins-stated,omitempty"`
|
||||
// Counted says the pins count: a pins-stated later than made is applied.
|
||||
Counted bool `json:"counted"`
|
||||
// Fill is the state's bytes against its cap.
|
||||
Fill meshcatalog.Fill `json:"fill"`
|
||||
}
|
||||
|
||||
// ReadPins reads the state pins and the revisions kept, for indexer_pins.
|
||||
func ReadPins(ctx context.Context, pins Store, o Objects) (PinsAnswer, error) {
|
||||
ix := &Indexer{Pins: pins, Revisions: o}
|
||||
var out PinsAnswer
|
||||
names, err := o.Names(ctx)
|
||||
if err != nil {
|
||||
return out, fmt.Errorf("the catalog graph revisions cannot be read: %w", err)
|
||||
}
|
||||
current := CurrentRevisions(names)
|
||||
pinnedObjects := map[string]bool{}
|
||||
pinKeys, err := pins.Keys(ctx)
|
||||
if err != nil {
|
||||
return out, fmt.Errorf("the state %s cannot be read: %w", PinsState, err)
|
||||
}
|
||||
slices.Sort(pinKeys)
|
||||
var bytes int64
|
||||
for _, k := range pinKeys {
|
||||
v, _, found, err := pins.Get(ctx, k)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
bytes += int64(len(k) + len(v))
|
||||
if object, ok := pinObject(k); ok {
|
||||
var pv PinValue
|
||||
if err := json.Unmarshal(v, &pv); err != nil {
|
||||
return out, fmt.Errorf("%s.%s: %w", PinsState, k, err)
|
||||
}
|
||||
why, cur := current[object]
|
||||
p := PinnedRevision{Revision: k, Object: object, Records: pv.Records, Missing: pv.Missing, Kept: slices.Contains(names, object), Current: cur, Why: why}
|
||||
if len(pv.Records) > 0 {
|
||||
pinnedObjects[object] = true
|
||||
}
|
||||
out.Pins = append(out.Pins, p)
|
||||
}
|
||||
}
|
||||
for _, n := range names {
|
||||
if _, _, ok := RevisionGroup(n); ok && !pinnedObjects[n] {
|
||||
out.Unpinned = append(out.Unpinned, n)
|
||||
}
|
||||
}
|
||||
if s, ok, err := ix.pinsStamp(ctx, PinsMadeKey); err == nil && ok {
|
||||
out.Made = &s.At
|
||||
}
|
||||
if s, ok, err := ix.pinsStamp(ctx, PinsStatedKey); err == nil && ok {
|
||||
out.Stated = &s.At
|
||||
}
|
||||
out.Counted, _ = ix.pinsCounted(ctx)
|
||||
out.Fill = meshcatalog.Fill{Bytes: bytes, Cap: PinsCapMiB << 20, Store: PinsState}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// PinsTools are indexer_pins (class read) and indexer_unpin (class act: an ask in takeins the indexer, the state's owner, applies;
|
||||
// record says the hand act in the hand-act log before it is asked).
|
||||
func PinsTools(pins Store, o Objects, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(what, why string) error) []stdio.Tool {
|
||||
return []stdio.Tool{
|
||||
{Name: PinsTool,
|
||||
Description: "Every catalog graph revision a delivery record pins in the indexer's state pins (design 47 G §579): the revision's " +
|
||||
"key `<branch>.<revision>`, the records pinning it, whether it was missing when pinned (finding pin-missing), whether it is " +
|
||||
"kept and whether it is current and why; the kept revisions no record pins; when the state was made and when the newest " +
|
||||
"pins-stated was applied, and whether pins count yet (a pins-stated later than made); and the state's fill against its cap. " +
|
||||
"indexer_consumer reads the consumer the events come through; indexer_unpin removes one record's pin.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{}},
|
||||
Run: func(map[string]any) (any, error) {
|
||||
return ReadPins(context.Background(), pins, o)
|
||||
}},
|
||||
{Name: UnpinTool,
|
||||
Description: "Remove one record's pin from one revision of the indexer's state pins, for a freed that was lost; the removal rule " +
|
||||
"then applies (a revision no record pins and not current goes). A hand act, recorded through mesh-hand-acts before it is asked. " +
|
||||
"Clears the finding pin-missing of a missing revision left with no record. indexer_pins names the revisions and records.",
|
||||
Input: map[string]any{"type": "object", "required": []string{"revision", "record", "why"}, "properties": map[string]any{
|
||||
"revision": map[string]any{"type": "string", "description": "the revision's key `<branch>.<revision>`, as indexer_pins names it"},
|
||||
"record": map[string]any{"type": "string", "description": "the delivery record's id"},
|
||||
"why": map[string]any{"type": "string", "description": "why it is unpinned, in plain words"}}},
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
str := func(k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
|
||||
revision, rec, why := str("revision"), str("record"), str("why")
|
||||
if _, _, ok := ParsePinKey(revision); !ok || rec == "" || why == "" {
|
||||
return nil, errors.New("name the revision as indexer_pins names it (revision) and the record (record) and say why (why): an unpin is a hand act, recorded through mesh-hand-acts")
|
||||
}
|
||||
data, err := json.Marshal(TakeInAsk{V: 1, UnpinRevision: revision, UnpinRecord: rec, Why: why, By: stdio.CallerOf(args), At: now().UTC()})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := record("unpinned the record "+rec+" from the catalog graph revision "+revision+" by hand", why); err != nil {
|
||||
return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was unpinned: %w", err)
|
||||
}
|
||||
a, err := ask("unpin."+keys.EscapeName(revision)+"."+keys.EscapeName(rec), data)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a.Result, nil
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
// ConsumerTool is indexer_consumer (class read): each durable consumer of the indexer as the process keeps it in the state consumers
|
||||
// (process.ConsumerLedger, A-176): its stream, name, acknowledged sequence, the messages pending, the last event applied with its
|
||||
// `at`, and when the oldest message delivered and not yet applied was delivered.
|
||||
const ConsumerTool = "indexer_consumer"
|
||||
|
||||
// ConsumersAnswer is what indexer_consumer answers.
|
||||
type ConsumersAnswer struct {
|
||||
Consumers []process.ConsumerFacts `json:"consumers"`
|
||||
}
|
||||
|
||||
// ReadConsumers reads the state consumers.
|
||||
func ReadConsumers(ctx context.Context, st Store) (ConsumersAnswer, error) {
|
||||
out := ConsumersAnswer{Consumers: []process.ConsumerFacts{}}
|
||||
ks, err := st.Keys(ctx)
|
||||
if err != nil {
|
||||
return out, fmt.Errorf("the state %s cannot be read: %w", process.ConsumersState, err)
|
||||
}
|
||||
slices.Sort(ks)
|
||||
for _, k := range ks {
|
||||
v, _, found, err := st.Get(ctx, k)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
var f process.ConsumerFacts
|
||||
if err := json.Unmarshal(v, &f); err != nil {
|
||||
return out, fmt.Errorf("%s.%s: %w", process.ConsumersState, k, err)
|
||||
}
|
||||
out.Consumers = append(out.Consumers, f)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ConsumerTools is indexer_consumer.
|
||||
func ConsumerTools(st Store) []stdio.Tool {
|
||||
return []stdio.Tool{{Name: ConsumerTool,
|
||||
Description: "Each durable consumer of the indexer, as its process keeps it in the state consumers after each event it applies: its stream " +
|
||||
"and name, the acknowledged sequence, the messages pending, the last event applied with its time, and when the oldest message delivered " +
|
||||
"and not yet applied was delivered. The pins events (mesh-delivery's pinned, freed, heads-replaced and pins-stated) come through it; " +
|
||||
"indexer_pins reads what they kept.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{}},
|
||||
Run: func(map[string]any) (any, error) { return ReadConsumers(context.Background(), st) }}}
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
|
||||
)
|
||||
|
||||
// Design 47 G §579, A-169, A-171: the indexer keeps the records pinning each catalog graph revision, counts a record once, and
|
||||
// removes a revision only while its consumer is caught up and after a pins-stated later than made: one two records pin stays
|
||||
// until both free it, one neither pinned nor current goes, a revision pinned and not held is said, and unpin clears it.
|
||||
func TestTheIndexerRemovesARevisionOnlyWhenNoRecordPinsItAndItIsNotCurrent(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
clock := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC)
|
||||
pending := uint64(0)
|
||||
ix := &Indexer{Revisions: newMemObjects(1 << 20), Pins: newMem(), Findings: newFindings(), Now: func() time.Time { return clock },
|
||||
PinsPending: func(context.Context) (uint64, error) { return pending, nil }}
|
||||
reader := meshcatalog.CatalogEntry{Module: "reader", Repository: "novox/r", Manifest: map[string]any{}}
|
||||
for _, commit := range []string{"c1", "c2", "c3"} {
|
||||
e := reader
|
||||
e.Commit = commit
|
||||
if err := ix.KeepBranchEntries(ctx, "main", 5, map[string]meshcatalog.CatalogEntry{"reader": e}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
kept := func() []string {
|
||||
n, _ := ix.Revisions.Names(ctx)
|
||||
return slices.DeleteFunc(n, func(s string) bool { _, _, ok := RevisionGroup(s); return !ok })
|
||||
}
|
||||
say := func(subject string, body any) {
|
||||
t.Helper()
|
||||
raw, _ := json.Marshal(body)
|
||||
if err := ix.HeardPins(ctx, subject, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
pins := func(record string, revisions ...int64) meshdelivery.Pins {
|
||||
p := meshdelivery.Pins{Record: record, Milestone: "47/G"}
|
||||
for _, r := range revisions {
|
||||
p.Catalog = append(p.Catalog, meshdelivery.PinsCatalogItem{Branch: "main", Revision: r})
|
||||
}
|
||||
return p
|
||||
}
|
||||
want := func(what string, names ...string) {
|
||||
t.Helper()
|
||||
if got := kept(); !slices.Equal(got, names) {
|
||||
t.Errorf("%s: kept %v, want %v", what, got, names)
|
||||
}
|
||||
}
|
||||
say(meshdelivery.SubjectPinned, pins("r1", 1))
|
||||
say(meshdelivery.SubjectPinned, pins("r1", 1)) // heard twice: pins once
|
||||
say(meshdelivery.SubjectPinned, pins("r2", 1))
|
||||
ix.JudgePins(ctx)
|
||||
want("before a pins-stated later than made is applied", "trunk.1", "trunk.2", "trunk.3")
|
||||
clock = clock.Add(time.Second)
|
||||
say(meshdelivery.SubjectPinsStated, meshdelivery.PinsStatedEvent{At: clock})
|
||||
pending = 3
|
||||
ix.JudgePins(ctx)
|
||||
want("while messages of the consumer are pending", "trunk.1", "trunk.2", "trunk.3")
|
||||
pending = 0
|
||||
ix.JudgePins(ctx)
|
||||
want("pinned by two records, the middle one neither pinned nor current", "trunk.1", "trunk.3")
|
||||
say(meshdelivery.SubjectFreed, pins("r1", 1))
|
||||
ix.JudgePins(ctx)
|
||||
want("one of two records freed it", "trunk.1", "trunk.3")
|
||||
say(meshdelivery.SubjectFreed, pins("r2", 1))
|
||||
ix.JudgePins(ctx)
|
||||
want("both freed it", "trunk.3")
|
||||
if keys, _ := ix.Pins.Keys(ctx); slices.Contains(keys, "main.1") {
|
||||
t.Errorf("the state holds %v, want no key of the removed revision", keys)
|
||||
}
|
||||
// A record pins a revision no longer held: its key says missing and the finding names it; unpin clears both.
|
||||
say(meshdelivery.SubjectPinned, pins("r3", 1))
|
||||
if f := ix.Findings.Held(); len(f) != 1 || f[0].Kind != PinMissingKind {
|
||||
t.Fatalf("findings %v, want one %s", f, PinMissingKind)
|
||||
}
|
||||
if r := ix.Unpin(ctx, "main.1", "r3", "operator", "freed was lost"); !r.Stored {
|
||||
t.Fatalf("unpin answered %q", r.Answer)
|
||||
}
|
||||
if f := ix.Findings.Held(); len(f) != 0 {
|
||||
t.Errorf("findings %v after unpin, want none", f)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAPinsStatedBeforeTheStateWasMadeCountsNothing(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
clock := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC)
|
||||
ix := &Indexer{Revisions: newMemObjects(1 << 20), Pins: newMem(), Now: func() time.Time { return clock },
|
||||
PinsPending: func(context.Context) (uint64, error) { return 0, nil }}
|
||||
raw, _ := json.Marshal(meshdelivery.PinsStatedEvent{At: clock.Add(-time.Hour)})
|
||||
if err := ix.HeardPins(ctx, meshdelivery.SubjectPinsStated, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ok, err := ix.pinsCounted(ctx); err != nil || ok {
|
||||
t.Errorf("counted %v, %v after a pins-stated older than made, want not counted", ok, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHeadsReplacedDropsTheCandidatesGraph(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
ix := &Indexer{Revisions: newMemObjects(1 << 20), Pins: newMem()}
|
||||
rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Kind: "candidate", Group: "feat-x", Heads: "0123456789ab", Revision: 1}, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, _ := json.Marshal(meshdelivery.HeadsReplacedEvent{Group: "feat-x", Heads: "0123456789ab", Repository: "novox/r", Number: 1})
|
||||
if err := ix.HeardPins(ctx, meshdelivery.SubjectHeadsReplaced, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n, _ := ix.Revisions.Names(ctx); len(n) != 0 {
|
||||
t.Errorf("the objects hold %v, want the candidate's graph dropped", n)
|
||||
}
|
||||
}
|
||||
@@ -73,7 +73,10 @@ type TakeInAsk struct {
|
||||
// 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"`
|
||||
// UnpinRevision and UnpinRecord ask one record's pin on a revision of the state pins to be removed (indexer_unpin).
|
||||
UnpinRevision string `json:"unpin-revision,omitempty"`
|
||||
UnpinRecord string `json:"unpin-record,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"`
|
||||
@@ -95,7 +98,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.NamedFingerprint == "" && 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.UnpinRevision == "" && a.Registry == nil {
|
||||
return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try")
|
||||
}
|
||||
return a, nil
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
@@ -131,7 +132,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
{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, 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},
|
||||
{Name: index.VersionsState, History: 1, MaxMiB: 64}, {Name: index.BuildEdgesState, History: 1}, {Name: index.InputsState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17
|
||||
{Name: index.VersionsState, History: 1, MaxMiB: 64}, {Name: index.BuildEdgesState, History: 1}, {Name: index.InputsState, History: 1}, {Name: index.PinsState, History: 1, MaxMiB: index.PinsCapMiB}, {Name: process.ConsumersState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
// Milestone 47/G W4.2 (design 47, Two graphs; ADR 0619 §2): the catalog graph revisions are objects in the old section,
|
||||
@@ -166,7 +167,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
// internal/broker/state.go:62-93).
|
||||
// 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-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-conditions.findings-asked", "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", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "mesh-wants.wanted", "node-build-go.step-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided", "seat:mesh-module-assigner.views.assignments"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
"mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pinned", "mesh-delivery.pins-stated", "mesh-delivery.source-moved", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "mesh-wants.wanted", "node-build-go.step-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided", "seat:mesh-module-assigner.views.assignments"}; !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" {
|
||||
@@ -201,8 +202,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
// 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, index.NamedDropTool, // design 47 G §675
|
||||
index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool, index.InputsTool) // design 47 G §667, §8
|
||||
if !reflect.DeepEqual(m.Tools, want) || len(want) != 47 || slices.Contains(m.Tools, "modules") {
|
||||
index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool, index.InputsTool, index.PinsTool, index.UnpinTool, index.ConsumerTool) // design 47 G §667, §8, §579
|
||||
if !reflect.DeepEqual(m.Tools, want) || len(want) != 50 || 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.
|
||||
|
||||
@@ -73,7 +73,10 @@
|
||||
"indexer_proposals",
|
||||
"indexer_proposal_drop",
|
||||
"indexer_seed_again",
|
||||
"indexer_inputs"
|
||||
"indexer_inputs",
|
||||
"indexer_pins",
|
||||
"indexer_unpin",
|
||||
"indexer_consumer"
|
||||
],
|
||||
"consumes": [
|
||||
"go-package-registry.published",
|
||||
@@ -92,6 +95,10 @@
|
||||
"mesh-delivery-checker.verdict",
|
||||
"mesh-delivery-forge-bridge.forge.**",
|
||||
"mesh-delivery.source-moved",
|
||||
"mesh-delivery.pinned",
|
||||
"mesh-delivery.freed",
|
||||
"mesh-delivery.heads-replaced",
|
||||
"mesh-delivery.pins-stated",
|
||||
"mesh-wants.wanted",
|
||||
"node-engine.running",
|
||||
"operator-channel.decided",
|
||||
@@ -123,7 +130,9 @@
|
||||
{ "name": "proposals", "history": 1, "max-mib": 4 },
|
||||
{ "name": "versions", "history": 1, "max-mib": 64 },
|
||||
{ "name": "build-edges", "history": 1 },
|
||||
{ "name": "inputs", "history": 1 }
|
||||
{ "name": "inputs", "history": 1 },
|
||||
{ "name": "pins", "history": 1, "max-mib": 4 },
|
||||
{ "name": "consumers", "history": 1 }
|
||||
],
|
||||
"streams": [
|
||||
{ "name": "registration-history", "events": ["registration.*"], "retention": "history", "max-mib": 64 },
|
||||
|
||||
Reference in New Issue
Block a user