The indexer's catch-up tool takes no arguments and is named beside its catalog graph revisions (milestone 47/G W4.14)
The node agent calls a restored part's catch-up tool with no arguments, so indexer_take_in with neither module nor repository takes in the head the gate's source names for every repository, as the first start's catch-up does, and keeps the view as the next revision when an entry moved. The manifest names it beside the catalog graph revisions, form 1 (design 47, Reverted). Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"
|
||||
|
||||
|
||||
@@ -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 }}
|
||||
|
||||
@@ -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 `<holder>.<state>`.
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user