diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index e1736091..d57fa4a3 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -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 diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index c44f37c8..4aa1dec0 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -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 `.` 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 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 1f4e3b58..8136c950 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -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 diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/pins.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/pins.go new file mode 100644 index 00000000..98e250f0 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/pins.go @@ -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) +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index 552b9afb..c61356e1 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -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 { diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index cce9eba5..f86ae3f7 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -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) diff --git a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go index cc0508d4..4134c2be 100644 --- a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go +++ b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go @@ -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) } diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index db632c15..3005971b 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -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...) } diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index b7261c76..37c90f43 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -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{}) diff --git a/modules/mesh-delivery-indexer/internal/index/pins.go b/modules/mesh-delivery-indexer/internal/index/pins.go new file mode 100644 index 00000000..f41237a4 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/pins.go @@ -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, `.` (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 `.`, 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 `.`, 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) }}} +} diff --git a/modules/mesh-delivery-indexer/internal/index/pins_test.go b/modules/mesh-delivery-indexer/internal/index/pins_test.go new file mode 100644 index 00000000..11437d42 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/pins_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index f007fddb..c75ce420 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -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 diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 853cc3db..7d03ef67 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -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. diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index b155f070..306fa37c 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -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 },