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 3126dbdd..2889ddc7 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 @@ -57,7 +57,10 @@ func Tools(ask asker, now func() time.Time, read stdio.KeptState, counts func() "source names for that repository, for a merge whose merged message the indexer never heard (the forge bridge has no " + "tool): a head whose trunk holds no module.json is taken in as a merge that leaves none (every module of the repository " + "is dropped, one another repository's trunk holds handed over from there first, listed in handed and dropped); a head " + - "that holds modules is answered \"take each module in by name\".", + "that holds modules is answered \"take each module in by name\". With neither module nor repository it is the indexer's " + + "catch-up tool, named in its manifest beside the catalog graph revisions: it takes in the head the gate's source names for " + + "every repository, as merges, and keeps the view as the next revision when an entry moved; the node agent calls it so " + + "after restoring them. Answers the repositories taken in and, in unread, those whose merged message was not read.", Input: map[string]any{"type": "object", "properties": map[string]any{ "module": map[string]any{"type": "string", "description": "the module's name, as the catalogue registers it"}, "repository": map[string]any{"type": "string", "description": "instead of module: the repository (owner/name) whose head is taken in"}, @@ -66,9 +69,12 @@ func Tools(ask asker, now func() time.Time, read stdio.KeptState, counts func() var key string var value []byte var err error - if str(args, "repository") != "" && str(args, "module") == "" { + switch { + case str(args, "repository") == "" && str(args, "module") == "": + key, value, err = index.NewCatchUp(str(args, "why"), stdio.CallerOf(args), now()) + case str(args, "repository") != "" && str(args, "module") == "": key, value, err = index.NewTakeInRepository(str(args, "repository"), str(args, "why"), stdio.CallerOf(args), now()) - } else { + default: key, value, err = index.NewTakeIn(str(args, "module"), str(args, "why"), stdio.CallerOf(args), now()) } if err != nil { 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 af5f2e8a..37822709 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -15,6 +15,7 @@ import ( type takerIn interface { TakeIn(ctx context.Context, module string) index.TakenIn TakeInRepository(ctx context.Context, repository string) index.TakenIn + CatchUp(ctx context.Context) index.TakenIn JudgeAgain(ctx context.Context, module, digest string) index.TakenIn SayAcceptedAgain(ctx context.Context, module string) index.TakenIn SeatDefinitionsAgain(ctx context.Context, module string, pull int, head string) index.TakenIn @@ -60,6 +61,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream } else if a.Pending != "" { asked, r = "hand take-in of what waits", ix.TakeInPendingByHand(ctx, a.Pending, a.By, a.Why) a.Module = a.Pending + } else if a.CatchUp { + asked, r = "catch-up of every repository", ix.CatchUp(ctx) + a.Module = r.Module } else if a.Repository != "" { asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository) a.Module = a.Repository diff --git a/modules/mesh-delivery-indexer/internal/index/registered.go b/modules/mesh-delivery-indexer/internal/index/registered.go index 2f30556c..5bf426d7 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered.go +++ b/modules/mesh-delivery-indexer/internal/index/registered.go @@ -220,6 +220,8 @@ type TakenIn struct { // and deleted. Handed []string `json:"handed,omitempty"` Dropped []string `json:"dropped,omitempty"` + // Unread are the repositories a catch-up (CatchUp) could not take in: their merged message was not read. + Unread []string `json:"unread,omitempty"` // Fingerprints are the modules whose fingerprint a recompute by hand moved (FingerprintsAgain), each old and new. Fingerprints []FingerprintMove `json:"fingerprints,omitempty"` } diff --git a/modules/mesh-delivery-indexer/internal/index/source.go b/modules/mesh-delivery-indexer/internal/index/source.go index 8ff4f934..702990d1 100644 --- a/modules/mesh-delivery-indexer/internal/index/source.go +++ b/modules/mesh-delivery-indexer/internal/index/source.go @@ -178,6 +178,42 @@ func (ix *Indexer) takeInHeads(ctx context.Context, heads map[string]string) []s return unread } +// CatchUp is the indexer's catch-up tool's work (indexer_take_in with no module and no repository, the catch-up its +// manifest names beside the catalog graph revisions): after a restore point of them is restored, the node agent calls it +// with no arguments, and the indexer takes in the head the gate's source names for every repository, as the first +// start's catch-up does, whatever revision the objects keep, then keeps the view as the next revision when an entry +// moved. Its answer names the repositories taken in and those whose merged message could not be read. +func (ix *Indexer) CatchUp(ctx context.Context) TakenIn { + out := TakenIn{Module: "catch-up"} + ix.mu.Lock() + heads := maps.Clone(ix.source.heads) + ix.mu.Unlock() + switch { + case len(heads) == 0: + out.Answer = "the gate's source names no head yet, so nothing is taken in; the next source revision is taken in as it comes" + return out + case ix.Modules == nil || ix.Message == nil: + out.Answer = "the indexer holds no view of the trunk or reads no merged message, so nothing is taken in" + return out + case !ix.catchingUp.CompareAndSwap(false, true): + out.Answer = "a catch-up runs already; its end keeps the revision" + return out + } + unread := ix.takeInHeads(ctx, heads) + ix.catchingUp.Store(false) + if err := ix.KeepView(ctx); err != nil { + out.Answer = "the repositories were taken in but the revision could not be kept: " + err.Error() + return out + } + out.Stored = len(unread) == 0 + out.Unread = unread + out.Answer = fmt.Sprintf("the heads the gate's source names for %d repositories taken in", len(heads)-len(unread)) + if len(unread) > 0 { + out.Answer += fmt.Sprintf("; %d not (%s): their merged message was not read, and the call may be made again", len(unread), strings.Join(unread, ", ")) + } + return out +} + // mergedAt is the forge bridge's merged message of a repository's head. func (ix *Indexer) mergedAt(ctx context.Context, repository, head string) (forgemsg.Forge, error) { data, err := ix.Message(ctx, repository, head, 0) diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 89505157..d6a9d524 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -32,6 +32,9 @@ type TakeInAsk struct { // Repository asks the head the gate's source names for a repository to be taken in (TakeInRepository) instead of a module. Repository string `json:"repository,omitempty"` Why string `json:"why,omitempty"` + // CatchUp asks the head the gate's source names for every repository to be taken in (CatchUp): the catch-up tool, + // indexer_take_in called with no module and no repository, as the node agent calls it after a restore. + CatchUp bool `json:"catch-up,omitempty"` // Judge asks the manifests to be judged again now (indexer_judge_again) instead of the module to be taken in. Judge bool `json:"judge,omitempty"` // Digest names, with Judge, the one manifest digest of the module to judge (indexer_judge with a digest). @@ -83,7 +86,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 { + if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp { return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try") } return a, nil @@ -114,6 +117,20 @@ func NewTakeInRepository(repository, why, by string, at time.Time) (string, []by return "repository." + key, data, err } +// CatchUpKey is the takeins key of a catch-up: one at a time, a second written over an unanswered first. +const CatchUpKey = "catch-up" + +// NewCatchUp is the value indexer_take_in writes when it names no module and no repository: the catch-up of every +// repository the gate's source names (CatchUp). +func NewCatchUp(why, by string, at time.Time) (string, []byte, error) { + why = strings.TrimSpace(why) + if why == "" { + why = "the catch-up after a restore" + } + data, err := json.Marshal(TakeInAsk{V: 1, CatchUp: true, Why: why, By: by, At: at.UTC()}) + return CatchUpKey, data, err +} + // TakeInMergeTool takes one merge in by hand, its merged message asked of the forge bridge (ADR 0633 §5, issue 1585). const TakeInMergeTool = "indexer_take_in_merge" diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go index c0eadfda..d20d2798 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go @@ -519,6 +519,24 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { // } else if raw, _, _, _ := view.Get(context.Background(), viewKey); !strings.Contains(string(raw), "a field the judge refuses") { t.Errorf("the view at a restart lost the verdict: %s", raw) } + // W4.14: the catch-up tool (indexer_take_in with no module and no repository, as the node agent calls it after a + // restore) takes in the head the source names for every repository again, whatever revision is kept, and keeps none + // when nothing moved; a repository whose message is not read is named in unread. + if key, value, err := NewCatchUp("", "node-engine@ace", time.Now()); err != nil || key != CatchUpKey { + t.Fatalf("NewCatchUp = %q (%v), want key %q", key, err, CatchUpKey) + } else if a, err := DecodeTakeIn(value); err != nil || !a.CatchUp { + t.Fatalf("a catch-up ask decodes to %+v (%v), want a catch-up", a, err) + } + asked := answers + if got := first.CatchUp(context.Background()); !got.Stored || answers != asked+1 || len(firstSaid) != 2 || len(got.Unread) != 0 { + t.Errorf("the catch-up answered %+v after %d asks, said %d revisions, want the one repository taken in, no new revision", got, answers-asked, len(firstSaid)) + } + first.Message = func(context.Context, string, string, int) ([]byte, error) { + return nil, errors.New("the forge bridge does not answer") + } + if got := first.CatchUp(context.Background()); got.Stored || !reflect.DeepEqual(got.Unread, []string{catalogueRepo}) { + t.Errorf("a catch-up whose message is not read answered %+v, want %s unread", got, catalogueRepo) + } // The tools bundle answers modules and graph from the newest trunk revision, their marks at its number; with none kept // it says it waits (view-waits), never an empty catalog. trunkStates := TrunkStates{Objects: first.Revisions, Under: func(string) views.Source { return nil }} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 209bfd62..25a07565 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -57,8 +57,10 @@ type manifest struct { RunsAs string `json:"runs-as"` Streams []stream `json:"streams"` Objects []struct { - Name string `json:"name"` - MaxMiB int `json:"max-mib"` + Name string `json:"name"` + MaxMiB int `json:"max-mib"` + Form int `json:"form"` + CatchUp string `json:"catch-up"` } `json:"objects"` Claims []struct { Name string `json:"name"` @@ -123,9 +125,11 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build 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, - // capped at 512 MiB, under the local name the indexer opens. - if len(m.Objects) != 1 || m.Objects[0].Name != index.CatalogRevisionsObjects || m.Objects[0].MaxMiB != 512 { - t.Errorf("objects = %+v, want %s capped at 512 MiB", m.Objects, index.CatalogRevisionsObjects) + // capped at 512 MiB, under the local name the indexer opens. W4.14 (design 47, Reverted): their form is 1, and the + // catch-up tool a restore of them is followed by is indexer_take_in, which the bundle serves. + if len(m.Objects) != 1 || m.Objects[0].Name != index.CatalogRevisionsObjects || m.Objects[0].MaxMiB != 512 || + m.Objects[0].Form != 1 || m.Objects[0].CatchUp != index.TakeInTool || !slices.Contains(m.Tools, index.TakeInTool) { + t.Errorf("objects = %+v, want %s capped at 512 MiB, form 1, catch-up %s (a tool of its own)", m.Objects, index.CatalogRevisionsObjects, index.TakeInTool) } // D-11 (ADR 0634 §1): the holders and buckets the indexer names (internal/index/names.go) are its manifest's: its own // name and states with their revisions, and the states it reads, each as `.`. diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index c3a0a3c7..ef26c524 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -98,7 +98,7 @@ { "name": "unjudged", "history": 1 } ], "objects": [ - { "name": "catalog-revisions", "max-mib": 512 } + { "name": "catalog-revisions", "max-mib": 512, "form": 1, "catch-up": "indexer_take_in" } ], "emits": [ "take-in.pending",