novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

A merge that leaves a repository no module never deletes a module another trunk holds, and indexer_take_in takes a repository in (issue 1367)

Browse Source 
dropRemoved hands the module over from the other trunk, read from the forge at the head the source names, so no wait in memory is needed. indexer_take_in with a repository takes in a head with no module.json as a merge that leaves none, and answers by name for a head with modules.
This commit is contained in:
jschoubben
2026-11-03 16:45:06 +01:00
parent 6bc16d74db
commit e1d0c9fe03
7 changed files with 228 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go
+17 -5
View File
@@ -48,12 +48,24 @@ func Tools(ask asker, now func() time.Time, read stdio.KeptState, counts func()
"indexer) is read again at that head and written when it changed. Answers stored, from which " +
"repository, commit and path, or why not (held already, no registration, the forge's error, no manifest there). " +
"The indexer reads such a module again by itself on the next source revision and merged message; this asks now. " +
"indexer_module (mesh-delivery-view) reads the result.",
Input: map[string]any{"type": "object", "required": []string{"module"}, "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "the module's name, as the catalogue registers it"},
"why": map[string]any{"type": "string", "description": "why it is taken in now, in plain words"}}},
"indexer_module (mesh-delivery-view) reads the result. With repository instead of module it takes in the head the gate's " +
"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\".",
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"},
"why": map[string]any{"type": "string", "description": "why it is taken in now, in plain words"}}},
Run: func(args map[string]any) (any, error) {
key, value, err := index.NewTakeIn(str(args, "module"), str(args, "why"), stdio.CallerOf(args), now())
var key string
var value []byte
var err error
if str(args, "repository") != "" && str(args, "module") == "" {
key, value, err = index.NewTakeInRepository(str(args, "repository"), str(args, "why"), stdio.CallerOf(args), now())
} else {
key, value, err = index.NewTakeIn(str(args, "module"), str(args, "why"), stdio.CallerOf(args), now())
}
if err != nil {
return nil, err
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go
+5 -1
View File
@@ -14,6 +14,7 @@ import (
// takerIn is the part of the indexer a takeins value asks through.
type takerIn interface {
TakeIn(ctx context.Context, module string) index.TakenIn
TakeInRepository(ctx context.Context, repository string) index.TakenIn
JudgeAgain(ctx context.Context, module string) index.TakenIn
}
@@ -33,7 +34,10 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
return
}
asked, r := "take-in", index.TakenIn{}
if a.Judge {
if a.Repository != "" {
asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository)
a.Module = a.Repository
} else if a.Judge {
asked, r = "judging again", ix.JudgeAgain(ctx, a.Module)
} else {
r = ix.TakeIn(ctx, a.Module)
modules/mesh-delivery-indexer/internal/index/modules.go
+12
View File
@@ -216,6 +216,7 @@ func (ix *Indexer) namedObjects(ctx context.Context, f delivery.Forge) (map[stri
// held(key, module) says whether the trunk holds it. A module of another repository, or of none known, is never touched.
func (ix *Indexer) dropRemoved(ctx context.Context, repository string, held map[string]delivery.Module, trunkHolds func(key string, m delivery.Module) bool) error {
var handed []string
elsewhere := map[string]delivery.Forge{}
for _, key := range slices.Sorted(maps.Keys(held)) {
cur := held[key]
if cur.Repository == "" || !delivery.SameRepository(cur.Repository, repository) || trunkHolds(key, cur) {
@@ -225,6 +226,12 @@ func (ix *Indexer) dropRemoved(ctx context.Context, repository string, held map[
handed = append(handed, key) // written in place by the repository it moves to, never deleted first
continue
}
// The wait of moves.go is memory, which a restart forgets (issue 1367): another repository's trunk that holds the
// module is asked, so the move survives it.
if f, ok := ix.heldByAnotherTrunk(ctx, key, repository); ok {
elsewhere[key] = f
continue
}
err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
if !found {
return nil, false, nil
@@ -244,6 +251,11 @@ func (ix *Indexer) dropRemoved(ctx context.Context, repository string, held map[
return err
}
}
for _, key := range slices.Sorted(maps.Keys(elsewhere)) {
if err := ix.handOverFrom(ctx, key, elsewhere[key]); err != nil {
return err
}
}
return nil
}
modules/mesh-delivery-indexer/internal/index/modules_test.go
+55 -7
View File
@@ -445,31 +445,76 @@ func (d *deletesSeen) Delete(ctx context.Context, key string, rev uint64) error
// holding repository's merge that drops it hands it over in place, never deleting its key; merged the other way round,
// the module is deleted and then written new.
func TestAModuleMovesBetweenRepositoriesOnlyWhenItsTrunkDropsIt(t *testing.T) {
const from, to = "novox/mesh-controller", "novox/mesh-networking"
const from, to, toHead = "novox/mesh-controller", "novox/mesh-networking", "bbbbbbbbbbbb"
held := mergedMsg(from, map[string]string{"modules/a": "t", "modules/b": "t"})
named := mergedMsg(to, map[string]string{"modules/a": "t2"})
dropped := mergedMsg(from, map[string]string{"modules/b": "t"})
gone := mergedMsg(from, nil) // issue 1367: a merge that leaves no module.json
gone.Trees, gone.Manifests = nil, nil
for _, tc := range []struct {
name string
merges []delivery.Forge
repository string // modules.a's repository at the end, "" for none
deletedKeyA bool
// Issue 1367: a fresh indexer, no wait in memory, the source naming heads; trunk is the module directories the forge
// reads at the head of `to`; takeIn is a repository taken in after the merges (indexer_take_in), whose answer must
// hold answer and be stored or not.
heads bool
trunk []string
takeIn string
answer string
stored bool
bGoneAtEnd bool
}{
{"named while its trunk holds it: left", []delivery.Forge{held, named}, from, false},
{"named, then dropped by its trunk: handed over in place", []delivery.Forge{held, named, dropped}, to, false},
{"dropped by its trunk, then named: deleted, then written new", []delivery.Forge{held, dropped, named}, to, true},
{"dropped by its trunk with no merge naming it: deleted", []delivery.Forge{held, dropped}, "", true},
{name: "named while its trunk holds it: left", merges: []delivery.Forge{held, named}, repository: from},
{name: "named, then dropped by its trunk: handed over in place", merges: []delivery.Forge{held, named, dropped}, repository: to},
{name: "dropped by its trunk, then named: deleted, then written new", merges: []delivery.Forge{held, dropped, named}, repository: to, deletedKeyA: true},
{name: "dropped by its trunk with no merge naming it: deleted", merges: []delivery.Forge{held, dropped}, deletedKeyA: true},
{name: "no module.json left, another trunk holds it, nothing waits in memory: written from that trunk, never deleted",
merges: []delivery.Forge{held, gone}, heads: true, trunk: []string{"modules/a", "modules/other"}, repository: to, bGoneAtEnd: true},
{name: "no module.json left, no other trunk holds it: deleted",
merges: []delivery.Forge{held, gone}, heads: true, deletedKeyA: true, bGoneAtEnd: true},
{name: "taken in by repository, a head with no module.json: handed over from the other trunk",
merges: []delivery.Forge{held}, heads: true, trunk: []string{"modules/a"}, takeIn: from, answer: "handed over", stored: true, repository: to, bGoneAtEnd: true},
{name: "taken in by repository, a head with modules: answered by name",
merges: []delivery.Forge{held}, heads: true, trunk: []string{"modules/a"}, takeIn: to, answer: "by name", repository: from},
{name: "taken in by repository, no head named: answered so",
merges: []delivery.Forge{held}, heads: true, takeIn: "novox/nobody", answer: "names no head", repository: from},
} {
t.Run(tc.name, func(t *testing.T) {
mods := &deletesSeen{memStore: newMem()}
ix := &Indexer{Store: newMem(), Modules: mods, Builds: NewBuilds()}
var logs []string
ix.Logf = func(format string, args ...any) { logs = append(logs, format) }
if tc.heads {
ix.source = sourceNow{rev: 7, heads: map[string]string{from: merge, to: toHead}}
ix.Trunk = func(_ context.Context, repository, sha string) ([]string, error) {
if repository == to && sha == toHead {
return tc.trunk, nil
}
if repository == from && tc.takeIn == from {
return nil, nil
}
return nil, nil
}
ix.Read = func(_ context.Context, repository, sha, dir string) (RepoRead, error) {
if repository != to || sha != toHead || dir != "modules/a" {
t.Errorf("read %s at %s %q, want only %s at %s modules/a", repository, sha, dir, to, toHead)
}
return RepoRead{Root: "r2", Trees: map[string]string{dir: "t2"}, Manifests: map[string]string{dir: `{"module":"a","version":"1"}`}}, nil
}
}
for _, f := range tc.merges {
if err := ix.Apply(context.Background(), f); err != nil {
t.Fatal(err)
}
}
if tc.takeIn != "" {
got := ix.TakeInRepository(context.Background(), tc.takeIn)
if got.Stored != tc.stored || !strings.Contains(got.Answer, tc.answer) {
t.Errorf("take-in answered %+v, want stored %v and %q", got, tc.stored, tc.answer)
}
}
m, ok := moduleOf(t, mods.memStore, "a")
if got := map[bool]string{true: m.Repository}[ok]; got != tc.repository {
t.Errorf("modules.a of %q, want %q (logs %v)", got, tc.repository, logs)
@@ -477,11 +522,14 @@ func TestAModuleMovesBetweenRepositoriesOnlyWhenItsTrunkDropsIt(t *testing.T) {
if ok && m.Path != "modules/a" {
t.Errorf("modules.a at %q, want modules/a", m.Path)
}
if ok && m.Repository == to && tc.heads && m.Commit != toHead {
t.Errorf("modules.a at %s, want the other trunk's head %s", m.Commit, toHead)
}
if deleted := slices.Contains(mods.deleted, "a"); deleted != tc.deletedKeyA {
t.Errorf("modules.a deleted %v, want %v", deleted, tc.deletedKeyA)
}
if _, ok := mods.vals["b"]; !ok {
t.Error("modules.b, which its trunk still holds, is gone")
if _, ok := mods.vals["b"]; ok == tc.bGoneAtEnd {
t.Errorf("modules.b held %v, want %v", ok, !tc.bGoneAtEnd)
}
if len(logs) == 0 {
t.Error("nothing was said")
modules/mesh-delivery-indexer/internal/index/moves.go
+61
View File
@@ -3,10 +3,66 @@ package index
import (
"context"
"fmt"
"maps"
"slices"
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// heldByAnotherTrunk reads, for a module the trunk of repository no longer holds, whether the trunk of another repository
// holds it at the head the latest source names: the module directories there are asked of the forge (Trunk) and the manifest of
// each directory named like the module is read (Read). It answers the merge to hand the module over with, or none: a forge
// that cannot say leaves the module to be dropped as it was, said (issue 1367).
func (ix *Indexer) heldByAnotherTrunk(ctx context.Context, module, repository string) (delivery.Forge, bool) {
if ix.Trunk == nil || ix.Read == nil {
return delivery.Forge{}, false
}
ix.mu.Lock()
heads := maps.Clone(ix.source.heads)
ix.mu.Unlock()
for _, other := range slices.Sorted(maps.Keys(heads)) {
head := heads[other]
if head == "" || delivery.SameRepository(other, repository) {
continue
}
dirs, err := ix.Trunk(ctx, other, head)
if err != nil {
ix.logf("%s at %s: the modules of the trunk cannot be read, so %s is not looked for there: %v", other, head, module, err)
continue
}
for _, d := range dirs {
dir := strings.Trim(d, "/")
if nameOfDirectory(dir) != module {
continue
}
read, err := ix.Read(ctx, other, head, dir)
if err != nil {
ix.logf("%s at %s: %q cannot be read, so %s is not handed over from it: %v", other, head, dir, module, err)
continue
}
raw := read.Manifests[dir]
if read.Trees[dir] == "" || manifestModule(raw) != module {
continue
}
f := delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: other, Merge: head, Root: read.Root,
Trees: map[string]string{dir: read.Trees[dir]}, Manifests: map[string]string{dir: raw}}
if l := read.Locks[dir]; len(l) > 0 {
f.Locks = map[string]map[string]string{dir: l}
}
if p := read.Packages[dir]; len(p) > 0 {
f.Packages = map[string][]string{dir: p}
}
ix.logf("%s at %s: the trunk holds the module %s at %q, so it is handed over from there and not deleted (issue 1367)", other, head, module, dir)
return f, true
}
}
return delivery.Forge{}, false
}
// nameOfDirectory is the last element of a module directory.
func nameOfDirectory(dir string) string { return dir[strings.LastIndex(dir, "/")+1:] }
// A module moves between repositories by the merges of the delivery group that moves it (novox/hq ADR 0420 clause 7, ADR
// 0501 clause 3; issue 1126: route-proxy leaves the controller's repository, which is archived). The merge of the
// repository it moves to names it while the indexer holds it of another: that merge leaves it, says so, and waits here.
@@ -57,6 +113,11 @@ func (ix *Indexer) handOver(ctx context.Context, key string) error {
if !ok {
return nil
}
return ix.handOverFrom(ctx, key, f)
}
// handOverFrom writes the module at key from the merge f, which names it, in place.
func (ix *Indexer) handOverFrom(ctx context.Context, key string, f delivery.Forge) error {
w, err := ix.applyMoving(ctx, f, false, map[string]bool{key: true})
if err != nil {
return fmt.Errorf("modules.%s, moving to %s: %w", key, f.Repository, err)
modules/mesh-delivery-indexer/internal/index/registered.go
+62
View File
@@ -213,6 +213,10 @@ type TakenIn struct {
Commit string `json:"commit,omitempty"`
Path string `json:"path,omitempty"`
Answer string `json:"answer"`
// Handed and Dropped are the modules a take-in of a repository (TakeInRepository) moved to another repository's trunk
// and deleted.
Handed []string `json:"handed,omitempty"`
Dropped []string `json:"dropped,omitempty"`
}
// TakeIn reads a registered module's manifest now, whatever was tried before, and writes it as a merge would (the tool
@@ -358,3 +362,61 @@ func (ix *Indexer) takeInHeld(ctx context.Context, module string) (TakenIn, bool
}
return out, true
}
// TakeInRepository takes in the head the gate's source names for a repository whose merged message the indexer never
// heard (the tool indexer_take_in with repository, issue 1367). A head whose trunk holds no module.json is applied as
// the merged message of a merge that leaves none: every module of the repository goes, one another trunk holds handed over
// from there first. A head whose trunk holds modules is answered "by name": each is taken in with its own name.
func (ix *Indexer) TakeInRepository(ctx context.Context, repository string) TakenIn {
out := TakenIn{Repository: repository}
ix.mu.Lock()
head := ""
for repo, h := range ix.source.heads {
if delivery.SameRepository(repo, repository) && h != "" {
head, out.Repository = h, repo
}
}
ix.mu.Unlock()
if head == "" || ix.Modules == nil || ix.Trunk == nil {
out.Answer = "the gate's source names no head of " + repository + ", or the indexer reads no forge or writes no modules"
return out
}
out.Commit = head
dirs, err := ix.Trunk(ctx, out.Repository, head)
if err != nil {
out.Answer = fmt.Sprintf("the modules of %s at %.12s cannot be read, so nothing is taken in: %v", repository, head, err)
return out
}
if len(dirs) > 0 {
out.Answer = fmt.Sprintf("%s at %.12s holds modules (%s): take each module in by name (module)", repository, head, strings.Join(dirs, ", "))
return out
}
before, err := ix.heldModules(ctx)
if err != nil {
out.Answer = "the modules state cannot be read: " + err.Error()
return out
}
f := delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: out.Repository, Merge: head}
if _, err := ix.ApplyMessage(ctx, f); err != nil {
out.Answer = fmt.Sprintf("the merge of %s at %.12s could not be taken in: %v", repository, head, err)
return out
}
after, err := ix.heldModules(ctx)
if err != nil {
out.Answer = "the modules state cannot be read: " + err.Error()
return out
}
for _, key := range slices.Sorted(maps.Keys(before)) {
if !delivery.SameRepository(before[key].Repository, repository) {
continue
}
if now, ok := after[key]; ok {
out.Handed = append(out.Handed, key+" to "+now.Repository+" at "+now.Commit)
} else {
out.Dropped = append(out.Dropped, key)
}
}
out.Stored = true
out.Answer = fmt.Sprintf("%s at %.12s holds no module.json: taken in as a merge that leaves none; handed over %v, dropped %v", repository, head, out.Handed, out.Dropped)
return out
}
modules/mesh-delivery-indexer/internal/index/takein.go
+16 -3
View File
@@ -27,7 +27,9 @@ const (
type TakeInAsk struct {
V int `json:"v"`
Module string `json:"module,omitempty"`
Why string `json:"why,omitempty"`
// 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"`
// 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"`
By string `json:"by,omitempty"`
@@ -48,8 +50,8 @@ 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 == "" {
return a, errors.New("a takeins value naming no module")
if a.Module == "" && a.Repository == "" {
return a, errors.New("a takeins value naming no module or repository")
}
return a, nil
}
@@ -67,3 +69,14 @@ func NewTakeIn(module, why, by string, at time.Time) (string, []byte, error) {
data, err := json.Marshal(TakeInAsk{V: 1, Module: module, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
return key, data, err
}
// NewTakeInRepository is the value indexer_take_in writes for a repository: its key and the value.
func NewTakeInRepository(repository, why, by string, at time.Time) (string, []byte, error) {
repository = strings.TrimSpace(repository)
key, err := delivery.RepositoryKey(repository)
if err != nil {
return "", nil, err
}
data, err := json.Marshal(TakeInAsk{V: 1, Repository: repository, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
return "repository." + key, data, err
}