From 08ed0cd7e0f9bd640a3931b6491dca67cdfae0d9 Mon Sep 17 00:00:00 2001 From: "samnayana@gmail.com" Date: Tue, 20 Oct 2026 20:41:26 +0530 Subject: [PATCH] feat(artifact)!: load versioned manifests as a set MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An artifact source publishes manifest_v1.json, manifest_v2.json, … and artifact.Set keeps one Registry per version. A published manifest and the files it lists never change: a change is a new file plus a new manifest. A flow records Set.Current() when it starts and resolves every artifact through Set.Registry(ctx, version), so a deploy never changes the templates of a running flow. - LoadSet probes manifest_v1.json upward until the loader reports one missing; any other error fails it. The highest version is current unless WithPinnedVersion names an older one, and with a pin only manifests up to it are read. - RegisterFromConfig rejects two rows with the same id, kind and version. LoadSet rejects an explicitly versioned row whose path differs between manifests, and GetExplicit serves such a row outside any flow. - Latest returns a manifest's unversioned row even when the manifest also lists explicit versions. BREAKING CHANGE: ManifestFilename gives way to ManifestPath(v), LoadManifest takes a version, and workflowdef.LoadVersion is removed. Co-Authored-By: Claude Opus 5.5 --- README.md | 8 +- artifact/README.md | 60 +++-- artifact/adapter/README.md | 6 +- artifact/adapter/workflowdef/workflowdef.go | 15 +- .../adapter/workflowdef/workflowdef_test.go | 21 -- artifact/manifest.go | 63 +++-- artifact/registry.go | 10 +- artifact/registry_test.go | 35 ++- artifact/set.go | 218 ++++++++++++++++ artifact/set_test.go | 238 ++++++++++++++++++ 10 files changed, 583 insertions(+), 91 deletions(-) create mode 100644 artifact/set.go create mode 100644 artifact/set_test.go diff --git a/README.md b/README.md index 58f42b5..44b250a 100644 --- a/README.md +++ b/README.md @@ -112,11 +112,11 @@ func Build(ctx context.Context, cfg *Config) (*App, error) { return nil, err } - // 2. Artifact registry — one loader is the single source of truth + // 2. Artifacts — one loader is the single source of truth. LoadSet reads + // configs/manifest_v1.json, manifest_v2.json, … artifactLoader, _ := local.New(local.Config{Root: "configs"}) - registry := artifact.NewRegistry(artifactLoader) - manifest, _ := artifact.LoadManifest(ctx, artifactLoader) // reads configs/manifest.json - artifact.RegisterFromConfig(registry, manifest) + artifacts, _ := artifact.LoadSet(ctx, artifactLoader) + registry := artifacts.CurrentRegistry() // 3. Payment paymentRepo := NewPaymentRepo(db) diff --git a/artifact/README.md b/artifact/README.md index 223884f..b9113ac 100644 --- a/artifact/README.md +++ b/artifact/README.md @@ -44,7 +44,11 @@ Return `artifact.ErrNotFound` when the resource is absent; return other errors o ### `Registry` -Holds registered loaders and a manifest table `(id, kind, version) → (loaderType, path)`. Created once at startup and shared read-only at runtime. +Resolves `(id, kind, version) → path` for **one manifest** and fetches through the single loader. Shared read-only at runtime. + +### `Set` + +Holds every published manifest of one source, one `Registry` per manifest version, and names the current one. A flow records `set.Current()` when it starts and resolves every artifact through `set.Registry(ctx, version)` until it ends, so a deploy never changes the templates of a running flow. See [Versioned manifests](#versioned-manifests). --- @@ -83,51 +87,61 @@ artifact.Latest[*EmailTemplate](ctx, reg, "welcome") ### 2. Wire up at startup ```go -// One loader is the single source of truth for this registry. +// One loader is the single source of truth for every manifest and artifact. loader, err := local.New(local.Config{Root: "/etc/configs"}) -reg := artifact.NewRegistry(loader) -// Register artifact rows directly in code... -reg.RegisterArtifact("welcome_email", "email_template", "v2", "emails/welcome.v2.json") +// Load manifest_v1.json, manifest_v2.json, … and pick the current version. +set, err := artifact.LoadSet(ctx, loader) // highest version +set, err := artifact.LoadSet(ctx, loader, artifact.WithPinnedVersion(4)) // or a pinned one +``` + +A bare `Registry` can still be built by hand, for tests: -// ...or load them from the manifest, fetched through the same loader. -cfg, err := artifact.LoadManifest(ctx, loader) -artifact.RegisterFromConfig(reg, cfg) +```go +reg := artifact.NewRegistry(loader) +reg.RegisterArtifact("welcome_email", "email_template", "", "emails/welcome.json") ``` -The loader is required — `NewRegistry(nil)` panics. `RegisterFromConfig` validates that each row has an `id`, `kind`, and `path`, so misconfiguration is caught at startup rather than at first access. +The loader is required — `NewRegistry(nil)` panics and `LoadSet(ctx, nil)` fails. `RegisterFromConfig` rejects a row without an `id`, `kind`, or `path`, and two rows with the same `id`, `kind` and `version`. ### 3. Fetch ```go -// Fetch the newest registered version +// Inside a flow: the version the flow recorded when it started. +reg, err := set.Registry(ctx, flow.ManifestVersion) tmpl, err := artifact.Latest[EmailTemplate](ctx, reg, "welcome_email") -// Fetch a specific pinned version (e.g. to resume a long-running workflow -// that must stay on the version it started with) -tmpl, err := artifact.Get[EmailTemplate](ctx, reg, "welcome_email", "v2") +// Outside any flow, for an artifact requested by explicit version (static data): +list, err := artifact.GetExplicit[StaticList](ctx, set, "ports", "2") ``` -Both calls: fetch bytes via the loader → call `Parse` → return typed value. +All calls: fetch bytes via the loader → call `Parse` → return typed value. Within one manifest an artifact without an explicit version has a single row, so `Latest` returns it. --- -## Manifest file +## Versioned manifests -The manifest is a single JSON file named `manifest.json`, resolved at the loader's -root (`artifact.ManifestFilename`). `LoadManifest` fetches it through the loader, -so the manifest lives alongside the artifacts it catalogs — one source for both. -There is no per-row `loader`: the registry has exactly one. +Manifests are `manifest_v1.json`, `manifest_v2.json`, … at the loader's root (`artifact.ManifestPath(v)`), alongside the artifacts they catalog. There is no per-row `loader`: the source has exactly one. ```json { "artifacts": [ - { "id": "welcome_email", "kind": "email_template", "version": "v2", "path": "emails/welcome.v2.json" }, - { "id": "import_clearance", "kind": "workflow", "version": "v10", "path": "workflows/import_clearance.v10.json" } + { "id": "welcome_email", "kind": "email_template", "path": "emails/welcome_v2.json" }, + { "id": "import_clearance", "kind": "workflow", "path": "workflows/import_clearance.json" }, + { "id": "ports", "kind": "static_data", "version": "2", "path": "static/ports_v2.json" } ] } ``` +The rules that make a manifest version a stable snapshot: + +- **Published files never change.** A change to an artifact is a new file (`welcome_v2.json`), and a new manifest, with the next version number, points the same ID at it. Every older manifest keeps resolving what it resolved when it was published. +- **Artifacts refer to each other by ID.** A task template names its workflow and render config by ID, and the manifest decides which file each ID is. Changing a render config needs a new manifest, not a new task template. +- **Versions are contiguous from 1.** `LoadSet` probes upward until the loader reports `ErrNotFound`, so a gap hides every later manifest. Any other load error fails `LoadSet`; for S3 this means the reader needs `s3:ListBucket`, or a missing key is reported as access denied. +- **An explicit `version` keeps its path.** A row with a `version` is fetched by that version outside any flow (`GetExplicit`), so every manifest that lists it must point at the same file. `LoadSet` rejects one that does not. +- **The current version** is the highest one, unless `WithPinnedVersion` names an older one for a rollback. With a pin, `LoadSet` reads only manifests up to it, so a broken newer manifest cannot block the rollback. +- **Within a manifest, the unversioned row wins.** `Latest` returns the row without a version when there is one, even if the manifest also lists explicit versions of the same ID. + --- ## Built-in loaders @@ -138,7 +152,7 @@ There is no per-row `loader`: the registry has exactly one. | `artifact/loaders/s3` | `s3.New(ctx, s3.Config{...})` | Reads objects from an S3 bucket; builds the client from `Region`/`Endpoint`/credentials (S3-compatible stores supported); `Prefix` scopes an in-bucket sub-folder for multi-deployment buckets | | `artifact/loaders/github` | `github.New(github.Config{...})` | Reads files from a GitHub repo at a pinned `Ref` over the Contents API (`net/http`); `BasePath` scopes a sub-directory for multi-deployment repos. Set `UseRawHost` to fetch public repos via `raw.githubusercontent.com` (no token, no REST rate limit) | -Both wrap their underlying "not found" errors as `artifact.ErrNotFound` so callers can use `errors.Is`. +All wrap their underlying "not found" errors as `artifact.ErrNotFound` so callers can use `errors.Is`. ### Selecting a loader from config @@ -190,6 +204,8 @@ m := testutil.MemLoader{ reg := artifact.NewRegistry(m) ``` +Put `manifest_v1.json` (and later versions) in a `MemLoader` to test against a `Set`. + --- ## Bridging to domain types diff --git a/artifact/adapter/README.md b/artifact/adapter/README.md index 9ce661b..3a4d26c 100644 --- a/artifact/adapter/README.md +++ b/artifact/adapter/README.md @@ -43,15 +43,11 @@ To make a type `T` (from package `dom`) loadable via the artifact registry, crea ``` 5. **Expose Fetch Helpers**: ```go + // reg is one manifest version's registry: set.Registry(ctx, version). func Load(ctx context.Context, reg *artifact.Registry, id string) (dom.T, error) { w, err := artifact.Latest[loadable](ctx, reg, id) return w.T, err } - - func LoadVersion(ctx context.Context, reg *artifact.Registry, id, version string) (dom.T, error) { - w, err := artifact.Get[loadable](ctx, reg, id, version) - return w.T, err - } ``` By following this pattern: diff --git a/artifact/adapter/workflowdef/workflowdef.go b/artifact/adapter/workflowdef/workflowdef.go index 9b458ab..ee03be5 100644 --- a/artifact/adapter/workflowdef/workflowdef.go +++ b/artifact/adapter/workflowdef/workflowdef.go @@ -17,7 +17,7 @@ const Kind artifact.Kind = "workflow" // loadable is the adapter: a local type (so we may legally define methods on it) // that embeds the pure DSL type and satisfies artifact.Artifact + artifact.Parser. -// Unexported — callers use Load/LoadVersion and never see it. +// Unexported — callers use Load and never see it. type loadable struct { engine.WorkflowDefinition } @@ -34,17 +34,10 @@ func (w *loadable) Parse(raw []byte) error { return nil } -// Load returns the newest version of a workflow definition. Callers get a plain -// engine.WorkflowDefinition back — they never see the adapter or know "artifact" -// was involved in fetching it. +// Load returns a workflow definition from reg, which is one manifest version's registry. Callers +// get a plain engine.WorkflowDefinition back — they never see the adapter or know "artifact" was +// involved in fetching it. func Load(ctx context.Context, reg *artifact.Registry, id string) (engine.WorkflowDefinition, error) { w, err := artifact.Latest[loadable](ctx, reg, id) return w.WorkflowDefinition, err } - -// LoadVersion returns a specific pinned version (e.g. to resume a running -// instance on the version it started with). -func LoadVersion(ctx context.Context, reg *artifact.Registry, id, version string) (engine.WorkflowDefinition, error) { - w, err := artifact.Get[loadable](ctx, reg, id, version) - return w.WorkflowDefinition, err -} diff --git a/artifact/adapter/workflowdef/workflowdef_test.go b/artifact/adapter/workflowdef/workflowdef_test.go index ca4f18d..4e6d942 100644 --- a/artifact/adapter/workflowdef/workflowdef_test.go +++ b/artifact/adapter/workflowdef/workflowdef_test.go @@ -36,27 +36,6 @@ func TestWorkflowDefAdapter(t *testing.T) { } }) - t.Run("LoadVersion returns pinned version", func(t *testing.T) { - m := testutil.MemLoader{ - "wf_v3.json": []byte(`{"id": "import_clearance", "name": "Clearance V3", "version": 3}`), - "wf_v4.json": []byte(`{"id": "import_clearance", "name": "Clearance V4", "version": 4}`), - } - reg := artifact.NewRegistry(m) - reg.RegisterArtifact("import_clearance", "workflow", "v3", "wf_v3.json") - reg.RegisterArtifact("import_clearance", "workflow", "v4", "wf_v4.json") - - def, err := workflowdef.LoadVersion(context.Background(), reg, "import_clearance", "v3") - if err != nil { - t.Fatalf("expected no error, got %v", err) - } - if def.Version != 3 { - t.Errorf("expected Version 3, got %d", def.Version) - } - if def.Name != "Clearance V3" { - t.Errorf("expected Name 'Clearance V3', got %q", def.Name) - } - }) - t.Run("Load returns latest version when multiple exist", func(t *testing.T) { m := testutil.MemLoader{ "wf_v1.json": []byte(`{"id": "import_clearance", "name": "Clearance V1", "version": 1}`), diff --git a/artifact/manifest.go b/artifact/manifest.go index 4ad36f5..00d299a 100644 --- a/artifact/manifest.go +++ b/artifact/manifest.go @@ -7,55 +7,74 @@ import ( "context" "encoding/json" "fmt" - "log/slog" ) -// ManifestFilename is the fixed name of the manifest, resolved at the loader's -// root. The manifest lives alongside the artifacts it catalogs, so a single -// loader resolves both; consumers do not choose its location. Point the loader's -// base (Root/Bucket/Repo) at wherever the manifest and artifacts reside. -const ManifestFilename = "manifest.json" +// ManifestPath returns the path of manifest version v, resolved at the loader's root: +// manifest_v1.json, manifest_v2.json, and so on. Manifests live alongside the artifacts they +// catalog, so one loader resolves both; consumers do not choose their location. Point the +// loader's base (Root/Bucket/Repo) at the directory that holds them. +// +// A manifest is immutable once published. A change to any artifact is a new file, listed by a +// new manifest with the next version number; see Set. +func ManifestPath(v int) string { + return fmt.Sprintf("manifest_v%d.json", v) +} type ManifestConfig struct { Artifacts []ManifestRow `json:"artifacts"` } +// ManifestRow names the file one artifact resolves to in a manifest. +// +// Version is optional. A row without one is the artifact for that manifest: the manifest +// itself is the version. A row with one is an artifact requested by explicit version (static +// data a form fetches by ID and version, for example); such a row must point at the same path +// in every manifest that lists it, so the version alone identifies the content. type ManifestRow struct { ID string `json:"id"` Kind Kind `json:"kind"` - Version string `json:"version"` // "" allowed for unversioned + Version string `json:"version"` Path string `json:"path"` } -// RegisterFromConfig applies every row via RegisterArtifact. Rows are validated -// so misconfiguration is caught at startup rather than at first access: id, kind, -// and path are all required (version may be "" for unversioned artifacts). There -// is no per-row loader — the registry has a single loader, so a row only names -// what to fetch, never how. +// RegisterFromConfig applies every row via RegisterArtifact. Rows are validated so +// misconfiguration is caught at startup rather than at first access: id, kind, and path are all +// required, and no two rows may share an id, kind and version. There is no per-row loader — the +// registry has a single loader, so a row only names what to fetch, never how. func RegisterFromConfig(r *Registry, cfg ManifestConfig) error { + seen := make(map[Key]int, len(cfg.Artifacts)) for i, row := range cfg.Artifacts { if row.ID == "" || row.Kind == "" || row.Path == "" { return fmt.Errorf("manifest row %d (%q/%q): id, kind, and path are all required", i, row.ID, row.Kind) } + k := Key{ID: row.ID, Kind: row.Kind, Version: row.Version} + if first, dup := seen[k]; dup { + return fmt.Errorf("manifest rows %d and %d both list %s/%s version %q", first, i, row.ID, row.Kind, row.Version) + } + seen[k] = i + } + for _, row := range cfg.Artifacts { r.RegisterArtifact(row.ID, row.Kind, row.Version, row.Path) } - slog.Info("artifact manifest registered", "count", len(cfg.Artifacts)) return nil } -// LoadManifest fetches and unmarshals the manifest from the artifact source -// itself, through the same loader that fetches the artifacts — so the catalog -// and its artifacts share one origin. The manifest always lives at -// ManifestFilename ("manifest.json") relative to the loader's root; its location -// is a fixed convention, not a parameter. -func LoadManifest(ctx context.Context, l Loader) (ManifestConfig, error) { +// LoadManifest fetches and unmarshals manifest version v from the artifact source itself, +// through the same loader that fetches the artifacts, so the catalog and its artifacts share +// one origin. A missing manifest returns an error matching ErrNotFound. +func LoadManifest(ctx context.Context, l Loader, v int) (ManifestConfig, error) { var cfg ManifestConfig - data, err := l.Load(ctx, ManifestFilename) + data, err := l.Load(ctx, ManifestPath(v)) if err != nil { - return cfg, fmt.Errorf("load manifest %q: %w", ManifestFilename, err) + return cfg, fmt.Errorf("load manifest %q: %w", ManifestPath(v), err) } + return parseManifest(v, data) +} + +func parseManifest(v int, data []byte) (ManifestConfig, error) { + var cfg ManifestConfig if err := json.Unmarshal(data, &cfg); err != nil { - return cfg, fmt.Errorf("unmarshal manifest %q: %w", ManifestFilename, err) + return ManifestConfig{}, fmt.Errorf("unmarshal manifest %q: %w", ManifestPath(v), err) } return cfg, nil } diff --git a/artifact/registry.go b/artifact/registry.go index 3fa899a..6d588c5 100644 --- a/artifact/registry.go +++ b/artifact/registry.go @@ -109,9 +109,10 @@ func Get[T Artifact](ctx context.Context, r *Registry, id, version string) (T, e return loadAndParse[T](ctx, r, e, id, kind, version) } -// Latest fetches and parses the newest version of an artifact, as type T. For an -// unversioned artifact (a single "" entry) it simply returns that entry, so it is -// also the natural "give me the current one" call for templates and schemas. +// Latest fetches and parses an artifact as type T: the row without an explicit version if +// the registry has one, otherwise the highest explicit version. In a manifest the row without +// a version is the artifact for that manifest, so Latest is the "give me this manifest's one" +// call for templates and schemas, even when the same manifest also lists explicit versions. func Latest[T Artifact](ctx context.Context, r *Registry, id string) (T, error) { var zero T kind := kindOf[T]() @@ -120,6 +121,9 @@ func Latest[T Artifact](ctx context.Context, r *Registry, id string) (T, error) if !ok || len(versions) == 0 { return zero, fmt.Errorf("%w: %s/%s", ErrNotFound, id, kind) } + if e, ok := versions[""]; ok { + return loadAndParse[T](ctx, r, e, id, kind, "") + } best, first := "", true for v := range versions { diff --git a/artifact/registry_test.go b/artifact/registry_test.go index b59e92b..c638549 100644 --- a/artifact/registry_test.go +++ b/artifact/registry_test.go @@ -268,19 +268,48 @@ func TestManifest(t *testing.T) { t.Errorf("expected required-fields error, got %v", err) } }) + + t.Run("RegisterFromConfig rejects two rows for one id, kind and version", func(t *testing.T) { + reg := artifact.NewRegistry(testutil.MemLoader{"a.json": []byte(`{"subject":"A"}`)}) + cfg := artifact.ManifestConfig{Artifacts: []artifact.ManifestRow{ + {ID: "form", Kind: "email", Path: "a.json"}, + {ID: "form", Kind: "email", Path: "b.json"}, + }} + err := artifact.RegisterFromConfig(reg, cfg) + if err == nil || !strings.Contains(err.Error(), "rows 0 and 1") { + t.Fatalf("expected a duplicate-row error naming both rows, got %v", err) + } + // Nothing is registered from a rejected manifest. + if _, err := artifact.Latest[fakeEmail](context.Background(), reg, "form"); !errors.Is(err, artifact.ErrNotFound) { + t.Errorf("expected ErrNotFound after a rejected manifest, got %v", err) + } + }) + + t.Run("RegisterFromConfig accepts one id under two kinds or two versions", func(t *testing.T) { + reg := artifact.NewRegistry(testutil.MemLoader{}) + cfg := artifact.ManifestConfig{Artifacts: []artifact.ManifestRow{ + {ID: "form", Kind: "email", Path: "a.json"}, + {ID: "form", Kind: "custom_ruleset", Path: "b.json"}, + {ID: "list", Kind: "email", Version: "1", Path: "c.json"}, + {ID: "list", Kind: "email", Version: "2", Path: "d.json"}, + }} + if err := artifact.RegisterFromConfig(reg, cfg); err != nil { + t.Fatalf("expected no error, got %v", err) + } + }) } func TestLoadManifest(t *testing.T) { t.Run("Load manifest from the source root", func(t *testing.T) { m := testutil.MemLoader{ - artifact.ManifestFilename: []byte(`{ + artifact.ManifestPath(1): []byte(`{ "artifacts": [ { "id": "test_id", "kind": "email", "version": "v1", "path": "path/to/email.json" } ] }`), } - cfg, err := artifact.LoadManifest(context.Background(), m) + cfg, err := artifact.LoadManifest(context.Background(), m, 1) if err != nil { t.Fatalf("expected no error, got %v", err) } @@ -290,7 +319,7 @@ func TestLoadManifest(t *testing.T) { }) t.Run("Missing manifest surfaces the loader error", func(t *testing.T) { - cfg, err := artifact.LoadManifest(context.Background(), testutil.MemLoader{}) + cfg, err := artifact.LoadManifest(context.Background(), testutil.MemLoader{}, 1) if !errors.Is(err, artifact.ErrNotFound) { t.Errorf("expected ErrNotFound, got %v", err) } diff --git a/artifact/set.go b/artifact/set.go new file mode 100644 index 0000000..0086cf3 --- /dev/null +++ b/artifact/set.go @@ -0,0 +1,218 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package artifact + +import ( + "context" + "errors" + "fmt" + "log/slog" + "slices" +) + +// ErrUnknownManifestVersion is returned when a manifest version is not positive or no manifest +// file exists for it. It does not match ErrNotFound: a flow pinned to a missing manifest is a +// deployment fault, not a missing artifact. +var ErrUnknownManifestVersion = errors.New("unknown manifest version") + +// Set holds every published manifest of one artifact source, each as its own Registry, and +// names the current one. +// +// Artifacts are immutable once a manifest lists them. A change is a new file, listed by a new +// manifest with the next version number, so an older manifest keeps resolving exactly what it +// resolved when it was published. A flow records the current version when it starts and +// resolves every artifact through Registry(version) until it ends, so a deploy never changes +// the templates of a flow already running. +// +// Manifests are found by probing manifest_v1.json, manifest_v2.json, … until one is missing, +// so versions must be contiguous. A Set does not change once loaded, so it is safe for +// concurrent use. +type Set struct { + loader Loader + + views map[int]*Registry + latest int + current int + + // explicit records the path each explicitly versioned row resolves to, across every loaded + // manifest, so a later manifest cannot repoint an (id, kind, version) at other content. + explicit map[Key]explicitRow +} + +type explicitRow struct { + path string + manifest int +} + +// SetOption configures LoadSet. +type SetOption func(*setOptions) + +type setOptions struct { + pin int +} + +// WithPinnedVersion makes manifest v the current one instead of the highest published version. +// It serves rollback, and lets a manifest be published before new flows use it. Zero means no +// pin. A pin to a version that does not exist fails LoadSet. +// +// With a pin, LoadSet reads only manifests 1 to v, so a newer manifest, even a malformed one, +// cannot stop a rollback to v. +func WithPinnedVersion(v int) SetOption { + return func(o *setOptions) { o.pin = v } +} + +// LoadSet loads every manifest the loader can see, starting at manifest_v1.json, and picks the +// current version. Loading stops at the first version whose manifest the loader reports as +// ErrNotFound, or at the pinned version; any other load error fails the call, so a transient +// fault never silently hides the newer manifests. Each manifest is checked as +// RegisterFromConfig checks it, and explicitly versioned rows are checked for a consistent path +// across manifests. Artifact files themselves are not fetched. +func LoadSet(ctx context.Context, loader Loader, opts ...SetOption) (*Set, error) { + if loader == nil { + return nil, errors.New("artifact: nil loader passed to LoadSet") + } + var o setOptions + for _, opt := range opts { + opt(&o) + } + if o.pin < 0 { + return nil, fmt.Errorf("%w: pinned version %d", ErrUnknownManifestVersion, o.pin) + } + + s := &Set{ + loader: loader, + views: make(map[int]*Registry), + explicit: make(map[Key]explicitRow), + } + cfgs, err := fetchManifests(ctx, loader, 1, o.pin) + if err != nil { + return nil, err + } + for i, cfg := range cfgs { + if err := s.add(i+1, cfg); err != nil { + return nil, err + } + } + if s.latest == 0 { + return nil, fmt.Errorf("%w: no %s found", ErrUnknownManifestVersion, ManifestPath(1)) + } + if o.pin > s.latest { + return nil, fmt.Errorf("%w: pinned version %d, highest published is %d", ErrUnknownManifestVersion, o.pin, s.latest) + } + s.current = s.latest + slog.InfoContext(ctx, "artifact manifests loaded", "latest", s.latest, "current", s.current, "pinned", o.pin != 0) + return s, nil +} + +// fetchManifests fetches manifests from version from upward, stopping at the first the loader +// reports missing or after version to (0: no limit). It returns those fetched before any other +// error, with that error. +func fetchManifests(ctx context.Context, loader Loader, from, to int) ([]ManifestConfig, error) { + var cfgs []ManifestConfig + for v := from; to == 0 || v <= to; v++ { + cfg, err := LoadManifest(ctx, loader, v) + if errors.Is(err, ErrNotFound) { + return cfgs, nil + } + if err != nil { + return cfgs, err + } + cfgs = append(cfgs, cfg) + } + return cfgs, nil +} + +// add registers manifest v. It runs only while LoadSet builds s. +func (s *Set) add(v int, cfg ManifestConfig) error { + reg := NewRegistry(s.loader) + if err := RegisterFromConfig(reg, cfg); err != nil { + return fmt.Errorf("%s: %w", ManifestPath(v), err) + } + // Check every explicit row before recording any, so a rejected manifest leaves no trace. + for _, row := range cfg.Artifacts { + if row.Version == "" { + continue + } + k := Key{ID: row.ID, Kind: row.Kind, Version: row.Version} + if prev, ok := s.explicit[k]; ok && prev.path != row.Path { + return fmt.Errorf("%s: %s/%s version %q points at %q, but %s points it at %q; an explicit version must keep its file", + ManifestPath(v), row.ID, row.Kind, row.Version, row.Path, ManifestPath(prev.manifest), prev.path) + } + } + for _, row := range cfg.Artifacts { + if row.Version == "" { + continue + } + k := Key{ID: row.ID, Kind: row.Kind, Version: row.Version} + if _, ok := s.explicit[k]; !ok { + s.explicit[k] = explicitRow{path: row.Path, manifest: v} + } + } + s.views[v] = reg + s.latest = v + return nil +} + +// Current returns the version new flows should pin: the highest published at load time, or the +// pinned one. A manifest published later becomes current when the process reloads the set. +func (s *Set) Current() int { + return s.current +} + +// Latest returns the highest manifest version loaded. +func (s *Set) Latest() int { + return s.latest +} + +// Versions returns every loaded manifest version, ascending. +func (s *Set) Versions() []int { + vs := make([]int, 0, len(s.views)) + for v := range s.views { + vs = append(vs, v) + } + slices.Sort(vs) + return vs +} + +// Registry returns the registry for manifest version v, or ErrUnknownManifestVersion if no +// loaded manifest has that version. +func (s *Set) Registry(_ context.Context, v int) (*Registry, error) { + reg, ok := s.views[v] + if !ok { + return nil, fmt.Errorf("%w: %d (highest published is %d)", ErrUnknownManifestVersion, v, s.latest) + } + return reg, nil +} + +// CurrentRegistry returns the registry for the current version. +func (s *Set) CurrentRegistry() *Registry { + return s.views[s.current] +} + +// GetExplicit fetches an explicitly versioned artifact without a manifest version, from +// whichever manifest lists it. LoadSet guarantees every manifest that lists an (id, kind, +// version) points it at the same path, so the result does not depend on which one is used. +// Use it for artifacts requested by explicit version outside any flow, such as static data. +func GetExplicit[T Artifact](ctx context.Context, s *Set, id, version string) (T, error) { + var zero T + kind := kindOf[T]() + if version == "" { + return zero, fmt.Errorf("%w: %s/%s needs an explicit version", ErrNotFound, id, kind) + } + key := Key{ID: id, Kind: kind, Version: version} + reg, ok := s.explicitRegistry(key) + if !ok { + return zero, fmt.Errorf("%w: %s/%s version %q", ErrNotFound, id, kind, version) + } + return Get[T](ctx, reg, id, version) +} + +// explicitRegistry returns a registry of a loaded manifest that lists key. +func (s *Set) explicitRegistry(key Key) (*Registry, bool) { + row, ok := s.explicit[key] + if !ok { + return nil, false + } + return s.views[row.manifest], true +} diff --git a/artifact/set_test.go b/artifact/set_test.go new file mode 100644 index 0000000..2edd6b6 --- /dev/null +++ b/artifact/set_test.go @@ -0,0 +1,238 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package artifact_test + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + + "github.com/OpenNSW/core/artifact" + "github.com/OpenNSW/core/artifact/testutil" +) + +func manifest(rows ...string) []byte { + return []byte(`{"artifacts":[` + strings.Join(rows, ",") + `]}`) +} + +func row(id, kind, version, path string) string { + return fmt.Sprintf(`{"id":%q,"kind":%q,"version":%q,"path":%q}`, id, kind, version, path) +} + +// twoVersions is a source where manifest v2 repoints "welcome" at a new file. +func twoVersions() testutil.MemLoader { + return testutil.MemLoader{ + "manifest_v1.json": manifest(row("welcome", "email", "", "welcome.json")), + "manifest_v2.json": manifest(row("welcome", "email", "", "welcome_v2.json")), + "welcome.json": []byte(`{"subject":"Hello"}`), + "welcome_v2.json": []byte(`{"subject":"Hello again"}`), + } +} + +func TestLoadSet(t *testing.T) { + ctx := context.Background() + + t.Run("loads every version and makes the highest current", func(t *testing.T) { + set, err := artifact.LoadSet(ctx, twoVersions()) + if err != nil { + t.Fatalf("LoadSet: %v", err) + } + if set.Current() != 2 || set.Latest() != 2 { + t.Errorf("current, latest = %d, %d; want 2, 2", set.Current(), set.Latest()) + } + if got := set.Versions(); len(got) != 2 || got[0] != 1 || got[1] != 2 { + t.Errorf("Versions() = %v, want [1 2]", got) + } + }) + + t.Run("each version resolves its own file", func(t *testing.T) { + set, err := artifact.LoadSet(ctx, twoVersions()) + if err != nil { + t.Fatalf("LoadSet: %v", err) + } + for v, want := range map[int]string{1: "Hello", 2: "Hello again"} { + reg, err := set.Registry(ctx, v) + if err != nil { + t.Fatalf("Registry(%d): %v", v, err) + } + got, err := artifact.Latest[fakeEmail](ctx, reg, "welcome") + if err != nil { + t.Fatalf("v%d: %v", v, err) + } + if got.Subject != want { + t.Errorf("v%d subject = %q, want %q", v, got.Subject, want) + } + } + cur, err := artifact.Latest[fakeEmail](ctx, set.CurrentRegistry(), "welcome") + if err != nil || cur.Subject != "Hello again" { + t.Errorf("current registry resolved %q, %v; want the v2 file", cur.Subject, err) + } + }) + + t.Run("a pin makes an older version current", func(t *testing.T) { + set, err := artifact.LoadSet(ctx, twoVersions(), artifact.WithPinnedVersion(1)) + if err != nil { + t.Fatalf("LoadSet: %v", err) + } + // Manifests above the pin are not read. + if set.Current() != 1 || set.Latest() != 1 { + t.Errorf("current, latest = %d, %d; want 1, 1", set.Current(), set.Latest()) + } + }) + + t.Run("a pin rolls back past a broken newer manifest", func(t *testing.T) { + src := twoVersions() + src["manifest_v3.json"] = manifest(row("welcome", "email", "", "a.json"), row("welcome", "email", "", "b.json")) + if _, err := artifact.LoadSet(ctx, src); err == nil { + t.Fatal("without a pin the broken manifest_v3.json should fail LoadSet") + } + set, err := artifact.LoadSet(ctx, src, artifact.WithPinnedVersion(2)) + if err != nil { + t.Fatalf("pinned to 2: %v", err) + } + if set.Current() != 2 || set.Latest() != 2 { + t.Errorf("current, latest = %d, %d; want 2, 2", set.Current(), set.Latest()) + } + if _, err := set.Registry(ctx, 3); !errors.Is(err, artifact.ErrUnknownManifestVersion) { + t.Errorf("Registry(3): expected ErrUnknownManifestVersion, got %v", err) + } + }) + + for name, pin := range map[string]int{"above the highest": 3, "negative": -1} { + t.Run("a pin "+name+" fails", func(t *testing.T) { + _, err := artifact.LoadSet(ctx, twoVersions(), artifact.WithPinnedVersion(pin)) + if !errors.Is(err, artifact.ErrUnknownManifestVersion) { + t.Fatalf("expected ErrUnknownManifestVersion, got %v", err) + } + }) + } + + t.Run("no manifest_v1.json fails", func(t *testing.T) { + _, err := artifact.LoadSet(ctx, testutil.MemLoader{"manifest.json": manifest()}) + if !errors.Is(err, artifact.ErrUnknownManifestVersion) { + t.Fatalf("expected ErrUnknownManifestVersion, got %v", err) + } + }) + + t.Run("a load error other than not-found fails instead of ending the probe", func(t *testing.T) { + src := twoVersions() + boom := errors.New("connection reset") + loader := errorLoader(func(ctx context.Context, path string) ([]byte, error) { + if path == "manifest_v2.json" { + return nil, boom + } + return src.Load(ctx, path) + }) + if _, err := artifact.LoadSet(ctx, loader); !errors.Is(err, boom) { + t.Fatalf("expected the loader error, got %v", err) + } + }) + + t.Run("a manifest with duplicate rows fails", func(t *testing.T) { + src := twoVersions() + src["manifest_v2.json"] = manifest(row("welcome", "email", "", "a.json"), row("welcome", "email", "", "b.json")) + _, err := artifact.LoadSet(ctx, src) + if err == nil || !strings.Contains(err.Error(), "manifest_v2.json") { + t.Fatalf("expected an error naming manifest_v2.json, got %v", err) + } + }) + + t.Run("a malformed manifest fails", func(t *testing.T) { + src := twoVersions() + src["manifest_v2.json"] = []byte(`{`) + if _, err := artifact.LoadSet(ctx, src); err == nil { + t.Fatal("expected an error") + } + }) + + t.Run("an explicit version must keep its path across manifests", func(t *testing.T) { + src := testutil.MemLoader{ + "manifest_v1.json": manifest(row("ports", "static_data", "1", "ports.json")), + "manifest_v2.json": manifest(row("ports", "static_data", "1", "ports_moved.json")), + } + _, err := artifact.LoadSet(ctx, src) + if err == nil || !strings.Contains(err.Error(), "must keep its file") { + t.Fatalf("expected an explicit-version path error, got %v", err) + } + + src["manifest_v2.json"] = manifest(row("ports", "static_data", "1", "ports.json"), row("ports", "static_data", "2", "ports_v2.json")) + if _, err := artifact.LoadSet(ctx, src); err != nil { + t.Fatalf("same path and a new version should load, got %v", err) + } + }) + + t.Run("nil loader fails", func(t *testing.T) { + if _, err := artifact.LoadSet(ctx, nil); err == nil { + t.Fatal("expected an error") + } + }) +} + +func TestSetRegistry(t *testing.T) { + ctx := context.Background() + + t.Run("rejects versions that cannot exist", func(t *testing.T) { + set, err := artifact.LoadSet(ctx, twoVersions()) + if err != nil { + t.Fatalf("LoadSet: %v", err) + } + for _, v := range []int{0, -1, 3} { + if _, err := set.Registry(ctx, v); !errors.Is(err, artifact.ErrUnknownManifestVersion) { + t.Errorf("Registry(%d): expected ErrUnknownManifestVersion, got %v", v, err) + } + if _, err := set.Registry(ctx, v); errors.Is(err, artifact.ErrNotFound) { + t.Errorf("Registry(%d): an unknown version must not match ErrNotFound", v) + } + } + }) +} + +func TestGetExplicit(t *testing.T) { + ctx := context.Background() + src := testutil.MemLoader{ + "manifest_v1.json": manifest(row("ports", "email", "1", "ports.json")), + "manifest_v2.json": manifest(row("ports", "email", "1", "ports.json"), row("ports", "email", "2", "ports_v2.json")), + "ports.json": []byte(`{"subject":"one"}`), + "ports_v2.json": []byte(`{"subject":"two"}`), + } + set, err := artifact.LoadSet(ctx, src) + if err != nil { + t.Fatalf("LoadSet: %v", err) + } + for version, want := range map[string]string{"1": "one", "2": "two"} { + got, err := artifact.GetExplicit[fakeEmail](ctx, set, "ports", version) + if err != nil || got.Subject != want { + t.Errorf("version %s: got %q, %v; want %q", version, got.Subject, err, want) + } + } + for _, version := range []string{"3", ""} { + if _, err := artifact.GetExplicit[fakeEmail](ctx, set, "ports", version); !errors.Is(err, artifact.ErrNotFound) { + t.Errorf("version %q: expected ErrNotFound, got %v", version, err) + } + } +} + +func TestLatestPrefersTheUnversionedRow(t *testing.T) { + ctx := context.Background() + set, err := artifact.LoadSet(ctx, testutil.MemLoader{ + "manifest_v1.json": manifest(row("welcome", "email", "", "current.json"), row("welcome", "email", "1", "old.json")), + "current.json": []byte(`{"subject":"current"}`), + "old.json": []byte(`{"subject":"old"}`), + }) + if err != nil { + t.Fatalf("LoadSet: %v", err) + } + got, err := artifact.Latest[fakeEmail](ctx, set.CurrentRegistry(), "welcome") + if err != nil || got.Subject != "current" { + t.Fatalf("Latest = %q, %v; want the unversioned row", got.Subject, err) + } + old, err := artifact.GetExplicit[fakeEmail](ctx, set, "welcome", "1") + if err != nil || old.Subject != "old" { + t.Fatalf("GetExplicit(1) = %q, %v; want the explicit row", old.Subject, err) + } +} + +// A version only a manifest published after the set loaded lists is found by probing.