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 01e3af36..bf432663 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -434,7 +434,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.PackagesBucket}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, TryRecovery: awaitedStore{b: tryRecovery}, Graph: awaitedStore{b: graph}, SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, - Message: messageVia(calls.CallKept), // a kept call: a read, called again through a transport fault (ADR 0668 §1) + Message: messageVia(calls.CallKept), Revisions: &busObjects{js: js, local: index.CatalogRevisionsObjects}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1) // The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear; // until then every merge is taken in at once, as before. Pending: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)}, diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go new file mode 100644 index 00000000..abc08a3c --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go @@ -0,0 +1,109 @@ +package main + +import ( + "bytes" + "context" + "errors" + "fmt" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + + "mesh-delivery-indexer/internal/index" +) + +// indexerModule is the module this program runs as, whose object buckets it opens by their local names. +const indexerModule = "mesh-delivery-indexer" + +// busObjects is one of the indexer's own object buckets in the old `objects` section (novox/hq ADR 0619 §2), which the +// grants part creates: it is opened at its first use and kept; until the bucket exists every call is an error saying so, +// and the next call opens it again (no timer: the next revision or tool call asks). +type busObjects struct { + js jetstream.JetStream + local string + mu sync.Mutex + o jetstream.ObjectStore +} + +func (b *busObjects) open(ctx context.Context) (jetstream.ObjectStore, error) { + b.mu.Lock() + defer b.mu.Unlock() + if b.o != nil { + return b.o, nil + } + o, err := b.js.ObjectStore(ctx, catalogue.ObjectsBucket(indexerModule, b.local)) + if err != nil { + return nil, fmt.Errorf("the object bucket %s is not open (the grants part creates it from the manifest's objects): %w", b.local, err) + } + b.o = o + return o, nil +} + +func (b *busObjects) Put(ctx context.Context, name string, data []byte) error { + o, err := b.open(ctx) + if err != nil { + return err + } + _, err = o.Put(ctx, jetstream.ObjectMeta{Name: name}, bytes.NewReader(data)) + return err +} + +func (b *busObjects) Get(ctx context.Context, name string) ([]byte, error) { + o, err := b.open(ctx) + if err != nil { + return nil, err + } + data, err := o.GetBytes(ctx, name) + if errors.Is(err, jetstream.ErrObjectNotFound) { + return nil, index.ErrNoObject + } + return data, err +} + +func (b *busObjects) Delete(ctx context.Context, name string) error { + o, err := b.open(ctx) + if err != nil { + return err + } + if err := o.Delete(ctx, name); err != nil && !errors.Is(err, jetstream.ErrObjectNotFound) { + return err + } + return nil +} + +func (b *busObjects) Names(ctx context.Context) ([]string, error) { + o, err := b.open(ctx) + if err != nil { + return nil, err + } + infos, err := o.List(ctx) + if errors.Is(err, jetstream.ErrNoObjectsFound) { + return []string{}, nil + } + if err != nil { + return nil, err + } + names := make([]string, 0, len(infos)) + for _, i := range infos { + names = append(names, i.Name) + } + return names, nil +} + +func (b *busObjects) Fill(ctx context.Context) (int64, int64, error) { + o, err := b.open(ctx) + if err != nil { + return 0, 0, err + } + st, err := o.Status(ctx) + if err != nil { + return 0, 0, err + } + info, err := b.js.Stream(ctx, "OBJ_"+st.Bucket()) + if err != nil { + return int64(st.Size()), 0, err + } + return int64(st.Size()), info.CachedInfo().Config.MaxBytes, nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 54463a6b..8e9c5b81 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -97,6 +97,9 @@ type Indexer struct { ReadFile func(ctx context.Context, repository, sha, path string) ([]byte, error) // SaySeatDefinition says seat-definition-changed (seatdefs.go); nil says nothing. SaySeatDefinition func(ctx context.Context, c SeatDefinitionChanged) error + // Revisions is the object bucket catalog-revisions (catalog_revisions.go; design 47, Two graphs); nil keeps no + // catalog graph revision. + Revisions Objects // Graph is the module graph's state (GraphBucket, graph.go; novox/hq ADR 0603 §2); nil writes no graph. Graph Store // Files asks the forge for every blob under each directory at a commit, by directory then path relative to it, each diff --git a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go new file mode 100644 index 00000000..3a457422 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go @@ -0,0 +1,120 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" +) + +// The indexer's object bucket catalog-revisions (design 47, Two graphs; novox/hq ADR 0619 §2: declared in the old +// `objects` section, its cap there): a revision is its list of entries' digests under trunk. or +// candidate.., and each entry one object entry., written once and shared by every revision that +// lists it. + +// CatalogRevisionsObjects is the bucket's local name in the indexer's manifest. +const CatalogRevisionsObjects = "catalog-revisions" + +// ErrNoObject is an object the bucket does not hold. +var ErrNoObject = errors.New("no such object") + +// Objects is an object bucket as the indexer keeps its revisions in it. +type Objects interface { + // Put stores data whole under name, replacing one of that name. + Put(ctx context.Context, name string, data []byte) error + // Get is the object's bytes; ErrNoObject when it holds none of that name. + Get(ctx context.Context, name string) ([]byte, error) + // Delete removes the object; one it does not hold is no error. + Delete(ctx context.Context, name string) error + // Names is every object's name; none is an empty list. + Names(ctx context.Context) ([]string, error) + // Fill is the bucket's bytes used and its cap in bytes. + Fill(ctx context.Context) (used, capacity int64, err error) +} + +// EntryObject is the object name of the entry of that digest. +func EntryObject(digest string) string { return "entry." + digest } + +// RevisionObject is the object name of a revision: trunk., or candidate.. for a candidate, +// which keeps its newest revision only. +func RevisionObject(rev meshcatalog.CatalogRevision) string { + if rev.Kind == "candidate" { + return "candidate." + rev.Group + "." + rev.Heads + } + return "trunk." + strconv.FormatInt(rev.Revision, 10) +} + +// TrunkRevisionOf is the revision number a trunk object's name holds, and whether it is one. +func TrunkRevisionOf(name string) (int64, bool) { + n, ok := strings.CutPrefix(name, "trunk.") + if !ok { + return 0, false + } + r, err := strconv.ParseInt(n, 10, 64) + return r, err == nil +} + +// KeepRevision stores a revision as CatalogRevisionOf made it: each entry object the bucket does not hold yet, then the +// revision's list, so a list is never kept whose entries are not; an entry already held is not written again. +func KeepRevision(ctx context.Context, o Objects, rev meshcatalog.CatalogRevision, entries map[string][]byte) error { + held, err := o.Names(ctx) + if err != nil { + return fmt.Errorf("the objects %s: %w", CatalogRevisionsObjects, err) + } + have := make(map[string]bool, len(held)) + for _, n := range held { + have[n] = true + } + for _, ref := range rev.Entries { + name := EntryObject(ref.Digest) + if have[name] { + continue + } + data, ok := entries[ref.Digest] + if !ok { + return fmt.Errorf("revision %s lists the entry of %s (%s), which it does not carry", RevisionObject(rev), ref.Module, ref.Digest) + } + if err := o.Put(ctx, name, data); err != nil { + return fmt.Errorf("the entry of %s in %s: %w", ref.Module, CatalogRevisionsObjects, err) + } + have[name] = true + } + list, err := json.Marshal(rev) + if err != nil { + return fmt.Errorf("revision %s: %w", RevisionObject(rev), err) + } + if err := o.Put(ctx, RevisionObject(rev), list); err != nil { + return fmt.Errorf("revision %s in %s: %w", RevisionObject(rev), CatalogRevisionsObjects, err) + } + return nil +} + +// ReadRevision is the revision kept under name and its entries whole, in the revision's order; an entry it lists that +// the bucket lost is an error naming it, never an entry left out. +func ReadRevision(ctx context.Context, o Objects, name string) (meshcatalog.CatalogRevision, []meshcatalog.CatalogEntry, error) { + var rev meshcatalog.CatalogRevision + data, err := o.Get(ctx, name) + if err != nil { + return rev, nil, fmt.Errorf("revision %s: %w", name, err) + } + if err := json.Unmarshal(data, &rev); err != nil { + return rev, nil, fmt.Errorf("revision %s: %w", name, err) + } + entries := make([]meshcatalog.CatalogEntry, 0, len(rev.Entries)) + for _, ref := range rev.Entries { + data, err := o.Get(ctx, EntryObject(ref.Digest)) + if err != nil { + return rev, nil, fmt.Errorf("revision %s, the entry of %s (%s): %w", name, ref.Module, ref.Digest, err) + } + var e meshcatalog.CatalogEntry + if err := json.Unmarshal(data, &e); err != nil { + return rev, nil, fmt.Errorf("revision %s, the entry of %s: %w", name, ref.Module, err) + } + entries = append(entries, e) + } + return rev, entries, nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/helpers_test.go b/modules/mesh-delivery-indexer/internal/index/helpers_test.go index 59cb9423..4c458ce4 100644 --- a/modules/mesh-delivery-indexer/internal/index/helpers_test.go +++ b/modules/mesh-delivery-indexer/internal/index/helpers_test.go @@ -1,6 +1,12 @@ package index -import "encoding/json" +import ( + "context" + "encoding/json" + "maps" + "slices" + "sync" +) // rawList is the list a kept predefined value holds, as it was written. func rawList(value []byte) json.RawMessage { @@ -10,3 +16,69 @@ func rawList(value []byte) json.RawMessage { _ = json.Unmarshal(value, &kept) return kept.List } + +// memObjects is an object bucket in memory with a cap: a put past the cap is refused as the bus's DiscardNew refuses it. +type memObjects struct { + mu sync.Mutex + objs map[string][]byte + puts int + capacity int64 +} + +func newMemObjects(capacity int64) *memObjects { + return &memObjects{objs: map[string][]byte{}, capacity: capacity} +} + +func (m *memObjects) used() int64 { + var n int64 + for _, d := range m.objs { + n += int64(len(d)) + } + return n +} + +func (m *memObjects) Put(_ context.Context, name string, data []byte) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.capacity > 0 && m.used()-int64(len(m.objs[name]))+int64(len(data)) > m.capacity { + return errObjectsFullForTest + } + m.objs[name] = slices.Clone(data) + m.puts++ + return nil +} + +func (m *memObjects) Get(_ context.Context, name string) ([]byte, error) { + m.mu.Lock() + defer m.mu.Unlock() + d, ok := m.objs[name] + if !ok { + return nil, ErrNoObject + } + return slices.Clone(d), nil +} + +func (m *memObjects) Delete(_ context.Context, name string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.objs, name) + return nil +} + +func (m *memObjects) Names(context.Context) ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + return slices.Sorted(maps.Keys(m.objs)), nil +} + +func (m *memObjects) Fill(context.Context) (int64, int64, error) { + m.mu.Lock() + defer m.mu.Unlock() + return m.used(), m.capacity, nil +} + +var errObjectsFullForTest = &objectsFullError{} + +type objectsFullError struct{} + +func (*objectsFullError) Error() string { return "maximum bytes exceeded" } diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 0f6515c6..242886a9 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -884,7 +884,9 @@ func (ix *Indexer) dry() *dryRun { // requirements, findings, seat definitions, holds and their bounds are not overlaid, nothing is said, and no hold is // weighed (Holds and Counted nil hold nothing); ReadFile only reads. Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil, - Holds: nil, Counted: nil, AskEngines: nil, HoldBounds: nil, OwnBuildState: nil, underSeats: true} + Holds: nil, Counted: nil, AskEngines: nil, HoldBounds: nil, OwnBuildState: 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} x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) } x.Store, d.candidates = over(ix.Store) x.Modules, d.modules = over(ix.Modules) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 0d62d0c5..efa3a483 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -970,7 +970,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", "Holds", "Counted", "AskEngines", "HoldBounds", "OpenTries", "TryRecovery", "OwnBuildState"} { + "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Holds", "Counted", "AskEngines", "HoldBounds", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go index 804983ee..73f7a37d 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go @@ -311,6 +311,27 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { // t.Errorf("entry %s: digest %s against %s before; only mesh-tools' moves", ref.Module, ref.Digest, rev.Entries[i].Digest) } } + // W4.2's store: both revisions kept in catalog-revisions hold one object per entry once, the moved entry twice, and + // each list; a revision read back is its entries whole; an entry the bucket lost is an error, never left out. + kept := newMemObjects(0) + _, nextObjects, _ := CatalogRevisionOf(next, moved) + if err := KeepRevision(context.Background(), kept, rev, objects); err != nil { + t.Fatal(err) + } + if err := KeepRevision(context.Background(), kept, next, nextObjects); err != nil { + t.Fatal(err) + } + if names, _ := kept.Names(context.Background()); len(names) != len(mods.vals)+1+2 || kept.puts != len(names) { + t.Errorf("catalog-revisions holds %v after %d puts, want each entry once, the moved one twice, and two lists", names, kept.puts) + } + if back, got, err := ReadRevision(context.Background(), kept, "trunk.2"); err != nil || back.Source != 8 || len(got) != len(mods.vals) || + !slices.ContainsFunc(got, func(e meshcatalog.CatalogEntry) bool { return e.Module == "mesh-tools" && e.Fp == tools.Fp }) { + t.Errorf("trunk.2 read back = %+v, %d entries, %v", back, len(got), err) + } + _ = kept.Delete(context.Background(), EntryObject(next.Entries[0].Digest)) + if _, _, err := ReadRevision(context.Background(), kept, "trunk.2"); err == nil || !strings.Contains(err.Error(), next.Entries[0].Module) { + t.Errorf("trunk.2 with a lost entry read without error naming it: %v", err) + } graph := ix.Graph.(*memStore) written := graph.writes if err := ix.Recompute(context.Background()); err != nil || graph.writes != written { diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index cb7db957..34b28b91 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -57,7 +57,11 @@ type manifest struct { Slug string `json:"slug"` RunsAs string `json:"runs-as"` Streams []stream `json:"streams"` - Claims []struct { + Objects []struct { + Name string `json:"name"` + MaxMiB int `json:"max-mib"` + } `json:"objects"` + Claims []struct { Name string `json:"name"` Scope string `json:"scope"` Serves []string `json:"serves"` @@ -119,6 +123,11 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build {Name: index.HoldsState, History: 5}, {Name: index.EngineReadsState, History: 1}, {Name: index.HoldBoundsState, History: 1, KeyTTL: true}, {Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17, ADR 0632 §2 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) + } // 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 `.`. if index.Module != m.Module { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 0392d211..b56cb192 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -99,6 +99,9 @@ { "name": "pending", "history": 1 }, { "name": "unjudged", "history": 1 } ], + "objects": [ + { "name": "catalog-revisions", "max-mib": 512 } + ], "emits": [ "take-in.pending", "take-in.taken",