amadan

carlos / platform Public

provision: RecordView keeps the record families and serves snapshots as a read-through store (#336)

RecordView follows the change index and re-reads only the account and app
scopes whose applied generation is behind the observed one, that are new,
or whose last re-read failed (retried on a 304 too). An unusable watch, a
view never built or a partial walk runs Rebuild, the backstop walk, which
reuses knownAccounts and listAppSubtree over a capturing store and sets
applied to the generations read before the walk. mu serializes both for
their whole duration; publication is an atomic pointer with a rising epoch.

ViewSnapshot.Store(live) answers the cached families (account listings and
manifests, app record listings and bodies, both manifests, the trash
marker) and sends everything else, and every key written through it, to
live. Failures are cached as the read errors themselves, so the unchanged
walk code over the snapshot reports partial/readFailed as it would live;
Complete is that walk's answer plus no outstanding scope.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1b265b1aa6e59431e46b5a8e9de3073b32114f52 parent 9770f3b
3 files changed, +1833 −0
  • internal/provision/recordview.go +575 −0
  • internal/provision/recordview_store.go +460 −0
  • internal/provision/recordview_test.go +798 −0
diff --git a/internal/provision/recordview.go b/internal/provision/recordview.go
new file mode 100644
index 000000000..65b0c403c
--- /dev/null
+++ b/internal/provision/recordview.go
@@ -0,0 +1,575 @@
+package provision
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "maps"
+ "os"
+ "slices"
+ "sort"
+ "strings"
+ "sync"
+ "sync/atomic"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// RecordView keeps the record families of the deployment bucket in memory
+// and follows the change index so a pass re-reads only the scopes that moved
+// (S3 change-index design §4). It holds, per account, the accounts/<a>/
+// listing and account.json; per app, the instances/ and sidecars/ listings
+// with every record body, the instances and sidecars manifests and the trash
+// marker. Consumers never read it directly: they take a Snapshot and run the
+// unchanged walk code (Reconcile, ReconcileSidecars, ListAllRecords*) over
+// Snapshot().Store(live).
+//
+// Two maps drive it (§4.2): the watch's observed generations and the view's
+// own applied ones. A scope whose applied generation is behind the observed
+// one, or that is new, or whose last re-read failed, is re-read on the next
+// Refresh, whether or not the index itself moved. Poll's moved list is never
+// used: several consumers share one watch, and whichever polls first takes it.
+type RecordView struct {
+ live release.Store
+ watch *release.ChangeWatch
+
+ // mu serializes Refresh and Rebuild for their whole duration, the backstop
+ // walk included, so a publication always reflects reads that began after
+ // the previous publication's (§4.3). It is a leaf: nothing else is taken
+ // while it is held (the watch's own locks are leaves below it). The fields
+ // after it are guarded by it.
+ mu sync.Mutex
+ applied map[string]uint64 // scope -> the generation the view's contents reflect
+ pending map[string]bool // scopes whose last read failed: re-read on every Refresh
+ rebuildDue bool // the last walk failed or could not list an account's apps
+ epoch uint64
+
+ snap atomic.Pointer[ViewSnapshot]
+}
+
+// recordViewRebuildHook, when set, runs inside Rebuild after the walk has
+// read everything and before it publishes, with mu held. Tests only.
+var recordViewRebuildHook func()
+
+// NewRecordView returns a view over live that follows watch. It has published
+// nothing: Snapshot is nil until the first Refresh or Rebuild succeeds.
+func NewRecordView(live release.Store, watch *release.ChangeWatch) *RecordView {
+ return &RecordView{live: live, watch: watch, applied: map[string]uint64{}, pending: map[string]bool{}}
+}
+
+// Snapshot is the latest publication, or nil before the first. It never
+// blocks on a refresh in progress.
+func (v *RecordView) Snapshot() *ViewSnapshot { return v.snap.Load() }
+
+// ScopeApp parses an app change scope, apps/<a>/<b>.
+func ScopeApp(scope string) (acct, app string, ok bool) {
+ rest, ok := strings.CutPrefix(scope, "apps/")
+ if !ok {
+ return "", "", false
+ }
+ acct, app, ok = strings.Cut(rest, "/")
+ if !ok || acct == "" || app == "" || strings.Contains(app, "/") {
+ return "", "", false
+ }
+ return acct, app, true
+}
+
+// scopeAccount parses an account change scope, accounts/<a>.
+func scopeAccount(scope string) (string, bool) {
+ a, ok := strings.CutPrefix(scope, "accounts/")
+ if !ok || a == "" || strings.Contains(a, "/") {
+ return "", false
+ }
+ return a, true
+}
+
+// viewScope reports whether the view keeps scope's contents. The control/
+// scopes belong to other readers.
+func viewScope(scope string) bool {
+ if _, _, ok := ScopeApp(scope); ok {
+ return true
+ }
+ _, ok := scopeAccount(scope)
+ return ok
+}
+
+// Refresh brings the view up to the change index (§4.2): one conditional GET
+// of the index, then a re-read of every scope whose applied generation is
+// behind the observed one, that is new, or whose last re-read failed. A view
+// that was never built, whose last walk was partial, or whose watch is
+// unusable (no index support, a denied or malformed index) runs Rebuild
+// instead, which is the walk every pass ran before the index (§3.4).
+//
+// A scope that fails to re-read keeps its applied generation, is retried on
+// the next Refresh even when the index answers 304, and is published with the
+// failure cached, so the snapshot is not Complete and a strict reader over it
+// fails as it would on the bucket. The error is only for a refresh that could
+// publish nothing (a failed walk, or ctx ending mid-read).
+func (v *RecordView) Refresh(ctx context.Context) error {
+ v.mu.Lock()
+ defer v.mu.Unlock()
+ prev := v.snap.Load()
+ if prev == nil || v.rebuildDue {
+ return v.rebuildLocked(ctx)
+ }
+ if _, err := v.watch.Poll(ctx); err != nil {
+ log.Printf("provision: record view: change index unusable (%v): walking", err)
+ return v.rebuildLocked(ctx)
+ }
+ obs, ok := v.watch.Observed()
+ if !ok {
+ return v.rebuildLocked(ctx)
+ }
+ var due []string
+ for scope, gen := range obs {
+ if have, ok := v.applied[scope]; viewScope(scope) && (!ok || have < gen) {
+ due = append(due, scope)
+ }
+ }
+ for scope := range v.pending {
+ if !slices.Contains(due, scope) {
+ due = append(due, scope)
+ }
+ }
+ if len(due) == 0 {
+ return nil
+ }
+ sort.Strings(due)
+
+ var mu sync.Mutex
+ results := make(map[string]scopeRead, len(due))
+ forEach(due, func(scope string) {
+ r := v.readScope(ctx, prev, scope)
+ mu.Lock()
+ results[scope] = r
+ mu.Unlock()
+ })
+ if err := ctx.Err(); err != nil {
+ return err // reads that died with the caller are not the bucket's answer
+ }
+ next := prev.clone()
+ for _, scope := range due {
+ r := results[scope]
+ if a, ok := scopeAccount(scope); ok {
+ next.setAccount(a, r.acct)
+ } else if a, b, ok := ScopeApp(scope); ok {
+ next.setApp(a, b, r.app)
+ }
+ if r.ok {
+ v.applied[scope] = obs[scope]
+ delete(v.pending, scope)
+ } else {
+ log.Printf("provision: record view: re-reading %s failed: retrying on the next refresh", scope)
+ v.pending[scope] = true
+ }
+ }
+ v.publish(next)
+ return nil
+}
+
+// Rebuild is the backstop (§4.4): under mu it polls the index, takes the
+// observed generations, walks every account and app, publishes, and sets
+// applied to the generations read BEFORE the walk, so a bump that lands
+// during the walk is re-read by the next Refresh, never skipped. Unchanged
+// bodies already in the view are kept by ETag rather than read again.
+//
+// A walk whose top-level listing fails publishes nothing and returns the
+// error; the previous snapshot stands and the next Refresh walks again.
+func (v *RecordView) Rebuild(ctx context.Context) error {
+ v.mu.Lock()
+ defer v.mu.Unlock()
+ return v.rebuildLocked(ctx)
+}
+
+func (v *RecordView) rebuildLocked(ctx context.Context) error {
+ if _, err := v.watch.Poll(ctx); err != nil {
+ log.Printf("provision: record view: change index unusable (%v): the walk records no generations", err)
+ }
+ obs, usable := v.watch.Observed()
+ sn, failed, err := v.walk(ctx, v.snap.Load())
+ if err == nil {
+ err = ctx.Err()
+ }
+ if err != nil {
+ v.rebuildDue = true
+ return fmt.Errorf("provision: record view: walk: %w", err)
+ }
+ if hook := recordViewRebuildHook; hook != nil {
+ hook()
+ }
+ v.applied = map[string]uint64{}
+ if usable {
+ for scope, gen := range obs {
+ if !viewScope(scope) || failed[scope] {
+ continue
+ }
+ if a, _, ok := ScopeApp(scope); ok && sn.dirErrs[a] != nil {
+ continue // its account's apps were never listed
+ }
+ v.applied[scope] = gen
+ }
+ }
+ v.pending = failed
+ v.rebuildDue = len(sn.dirErrs) > 0
+ v.publish(sn)
+ return nil
+}
+
+// publish stamps sn with the next epoch and its completeness and swaps it in.
+func (v *RecordView) publish(sn *ViewSnapshot) {
+ sn.Complete = len(v.pending) == 0 && !v.rebuildDue && walkComplete(sn)
+ v.epoch++
+ sn.Epoch = v.epoch
+ v.snap.Store(sn)
+}
+
+// walkComplete runs the record walk itself over the snapshot, answering from
+// the cache alone, and reports whether it would be partial: an account
+// manifest that will not read cleanly, or an app listing that failed. Those
+// are exactly what a strict reader refuses on the bucket.
+func walkComplete(sn *ViewSnapshot) bool {
+ ctx := context.Background()
+ s := sn.Store(uncachedStore{})
+ known, readFailed, err := knownAccounts(ctx, s)
+ if err != nil || readFailed {
+ return false
+ }
+ for _, sub := range []string{"instances/", "sidecars/"} {
+ if _, partial, err := listAppSubtree(ctx, s, known, sub); err != nil || partial {
+ return false
+ }
+ }
+ return true
+}
+
+// scopeRead is one scope's re-read: the new entry and whether every read in
+// it succeeded.
+type scopeRead struct {
+ acct *acctEntry
+ app *appEntry
+ ok bool
+}
+
+func (v *RecordView) readScope(ctx context.Context, prev *ViewSnapshot, scope string) scopeRead {
+ if a, ok := scopeAccount(scope); ok {
+ e, ok := v.readAccount(ctx, a, prev.accounts[a])
+ return scopeRead{acct: e, ok: ok}
+ }
+ a, b, _ := ScopeApp(scope)
+ p := release.AppPrefix(a, b)
+ inst, instErr := listETags(ctx, v.live, p+"instances/")
+ side, sideErr := listETags(ctx, v.live, p+"sidecars/")
+ e, ok := v.readApp(ctx, a, b, prev.apps[a][b], listing{inst, instErr}, listing{side, sideErr})
+ return scopeRead{app: e, ok: ok}
+}
+
+// readAccount re-reads accounts/<a>/: its listing, and account.json only
+// when its ETag moved.
+func (v *RecordView) readAccount(ctx context.Context, a string, prev *acctEntry) (*acctEntry, bool) {
+ key := accountManifestKey(a)
+ kes, err := listETags(ctx, v.live, "accounts/"+a+"/")
+ if err != nil {
+ return &acctEntry{listErr: err, manifest: object{err: err}}, false
+ }
+ e := &acctEntry{manifest: absent(key)}
+ i := slices.IndexFunc(kes, func(k release.KeyETag) bool { return k.Key == key })
+ if i >= 0 {
+ var was *object
+ if prev != nil {
+ was = &prev.manifest
+ }
+ e.manifest = v.fetch(ctx, key, kes[i].ETag, was)
+ if errors.Is(e.manifest.err, os.ErrNotExist) {
+ kes = slices.Delete(kes, i, i+1) // deleted since it was listed
+ }
+ }
+ e.keys = kes
+ return e, !e.manifest.failed()
+}
+
+// listing is one prefix's ETag listing, or why it failed.
+type listing struct {
+ kes []release.KeyETag
+ err error
+}
+
+// readApp fills one app's entry from its two listings: every listed record
+// body (kept from prev when the ETag is unchanged), then both manifests and
+// the trash marker by conditional GET.
+func (v *RecordView) readApp(ctx context.Context, a, b string, prev *appEntry, inst, side listing) (*appEntry, bool) {
+ e := &appEntry{instErr: inst.err, sideErr: side.err, objs: map[string]object{}}
+ ok := inst.err == nil && side.err == nil
+ read := func(kes []release.KeyETag) []release.KeyETag {
+ var kept []release.KeyETag
+ for _, ke := range kes {
+ o := v.fetch(ctx, ke.Key, ke.ETag, prev.object(ke.Key))
+ if errors.Is(o.err, os.ErrNotExist) {
+ continue // deleted since it was listed: the listing follows the bucket
+ }
+ if o.failed() {
+ ok = false
+ }
+ e.objs[ke.Key] = o
+ kept = append(kept, ke)
+ }
+ return kept
+ }
+ e.inst, e.side = read(inst.kes), read(side.kes)
+ for _, key := range []string{ManifestKey(a, b), SidecarManifestKey(a, b), release.AppTombstoneKey(a, b)} {
+ o := condGet(ctx, v.live, key, prev.object(key))
+ if o.failed() {
+ ok = false
+ }
+ e.objs[key] = o
+ }
+ return e, ok
+}
+
+// fetch reads key unless prev already holds the body the listing's ETag
+// names.
+func (v *RecordView) fetch(ctx context.Context, key, listed string, prev *object) object {
+ if prev != nil && prev.err == nil && listed != "" && prev.etag == listed {
+ return *prev
+ }
+ return condGet(ctx, v.live, key, prev)
+}
+
+// condGet reads key with a conditional GET against prev's ETag, so an
+// unchanged object costs a 304. Any error from the extension other than
+// not-found means "unknown" and falls back to a plain Get, as does a store
+// without it.
+func condGet(ctx context.Context, s release.Store, key string, prev *object) object {
+ etag := ""
+ if prev != nil && prev.err == nil {
+ etag = prev.etag
+ }
+ if g, ok := s.(release.ConditionalGetter); ok {
+ body, newETag, changed, err := g.GetIfChanged(ctx, key, etag)
+ switch {
+ case err == nil && !changed && etag != "":
+ return *prev
+ case err == nil && changed:
+ return object{body: body, etag: newETag}
+ case errors.Is(err, os.ErrNotExist):
+ return object{err: err}
+ }
+ }
+ body, newETag, err := s.Get(ctx, key)
+ return object{body: body, etag: newETag, err: err}
+}
+
+// listETags lists prefix with each key's ETag, sorted. A store without
+// ETagLister, or any error from it, falls back to List with no ETags, which
+// only costs the body reads the ETags would have saved.
+func listETags(ctx context.Context, s release.Store, prefix string) ([]release.KeyETag, error) {
+ var kes []release.KeyETag
+ l, ok := s.(release.ETagLister)
+ var err error
+ if ok {
+ kes, err = l.ListETags(ctx, prefix)
+ }
+ if !ok || err != nil {
+ keys, err := s.List(ctx, prefix)
+ if err != nil {
+ return nil, err
+ }
+ kes = make([]release.KeyETag, len(keys))
+ for i, k := range keys {
+ kes[i] = release.KeyETag{Key: k}
+ }
+ }
+ slices.SortFunc(kes, func(x, y release.KeyETag) int { return strings.Compare(x.Key, y.Key) })
+ return kes, nil
+}
+
+// walk reads the whole deployment with the record walk's own code, knownAccounts
+// and listAppSubtree, over a store that keeps what they read. Apps are walked
+// under every account directory, not only known ones, so an account that
+// becomes known later (restored, or materialized after its apps) already has
+// its apps in the view. failed holds every scope with a read that failed.
+func (v *RecordView) walk(ctx context.Context, prev *ViewSnapshot) (*ViewSnapshot, map[string]bool, error) {
+ w := &walkStore{Store: v.live, prev: prev, lists: map[string]listing{}, dirs: map[string]dirListing{}, gets: map[string]object{}}
+ known, _, err := knownAccounts(ctx, w)
+ if err != nil {
+ return nil, nil, err
+ }
+ every := maps.Clone(known)
+ bucket := release.HasBucketListDirs(v.live)
+ if bucket {
+ dirs, err := w.ListDirs(ctx, "apps/")
+ if err != nil {
+ return nil, nil, fmt.Errorf("list apps: %w", err)
+ }
+ for _, d := range dirs {
+ every[strings.TrimSuffix(strings.TrimPrefix(d, "apps/"), "/")] = true
+ }
+ }
+ for _, sub := range []string{"instances/", "sidecars/"} {
+ if _, _, err := listAppSubtree(ctx, w, every, sub); err != nil {
+ return nil, nil, fmt.Errorf("list apps: %w", err)
+ }
+ }
+
+ sn := &ViewSnapshot{accounts: map[string]*acctEntry{}, apps: map[string]map[string]*appEntry{}, dirErrs: map[string]error{}}
+ failed := map[string]bool{}
+ for _, ke := range w.lists["accounts/"].kes {
+ scope, ok := release.ChangeScope(ke.Key)
+ a, isAcct := scopeAccount(scope)
+ if !ok || !isAcct {
+ sn.strays = append(sn.strays, ke.Key)
+ continue
+ }
+ e := sn.accounts[a]
+ if e == nil {
+ e = &acctEntry{manifest: absent(accountManifestKey(a))}
+ sn.accounts[a] = e
+ }
+ e.keys = append(e.keys, ke)
+ }
+ for a, e := range sn.accounts {
+ key := accountManifestKey(a)
+ if !slices.ContainsFunc(e.keys, func(k release.KeyETag) bool { return k.Key == key }) {
+ continue
+ }
+ o, ok := w.gets[key]
+ if !ok { // knownAccounts reads every manifest it lists; this is belt and braces
+ o = condGet(ctx, v.live, key, nil)
+ }
+ e.manifest = o
+ if o.failed() {
+ failed["accounts/"+a] = true
+ }
+ }
+
+ type appListings struct{ inst, side listing }
+ found := map[[2]string]*appListings{}
+ at := func(a, b string) *appListings {
+ k := [2]string{a, b}
+ if found[k] == nil {
+ found[k] = &appListings{}
+ }
+ return found[k]
+ }
+ if bucket {
+ for _, d := range w.dirs["apps/"].dirs {
+ a := strings.TrimSuffix(strings.TrimPrefix(d, "apps/"), "/")
+ ad, ok := w.dirs[d]
+ if !ok {
+ continue
+ }
+ if ad.err != nil {
+ sn.dirErrs[a] = ad.err
+ continue
+ }
+ for _, p := range ad.dirs {
+ b := strings.TrimSuffix(strings.TrimPrefix(p, d), "/")
+ l := at(a, b)
+ l.inst, l.side = w.lists[p+"instances/"], w.lists[p+"sidecars/"]
+ }
+ }
+ } else {
+ for _, ke := range w.lists["apps/"].kes {
+ parts := strings.SplitN(ke.Key, "/", 4)
+ if len(parts) < 4 || parts[1] == "" || parts[2] == "" {
+ continue
+ }
+ l := at(parts[1], parts[2])
+ switch {
+ case strings.HasPrefix(parts[3], "instances/"):
+ l.inst.kes = append(l.inst.kes, ke)
+ case strings.HasPrefix(parts[3], "sidecars/"):
+ l.side.kes = append(l.side.kes, ke)
+ }
+ }
+ }
+ ids := make([]string, 0, len(found))
+ for k := range found {
+ ids = append(ids, "apps/"+k[0]+"/"+k[1])
+ }
+ var mu sync.Mutex
+ forEach(ids, func(scope string) {
+ a, b, _ := ScopeApp(scope)
+ l := found[[2]string{a, b}]
+ e, ok := v.readApp(ctx, a, b, prev.app(a, b), l.inst, l.side)
+ mu.Lock()
+ defer mu.Unlock()
+ sn.setApp(a, b, e)
+ if !ok {
+ failed[scope] = true
+ }
+ })
+ return sn, failed, nil
+}
+
+// walkStore is the live store as the backstop walk sees it: every listing
+// carries ETags and every read is conditional on what the previous snapshot
+// held, and both are kept for the snapshot the walk builds. Each listing is
+// read once per walk, however many times the walk code asks.
+type walkStore struct {
+ release.Store
+ prev *ViewSnapshot
+
+ mu sync.Mutex
+ lists map[string]listing
+ dirs map[string]dirListing
+ gets map[string]object
+}
+
+type dirListing struct {
+ dirs []string
+ err error
+}
+
+func (w *walkStore) BucketShapedListDirs() bool { return release.HasBucketListDirs(w.Store) }
+
+func (w *walkStore) List(ctx context.Context, prefix string) ([]string, error) {
+ w.mu.Lock()
+ l, ok := w.lists[prefix]
+ w.mu.Unlock()
+ if !ok {
+ kes, err := listETags(ctx, w.Store, prefix)
+ l = listing{kes, err}
+ w.mu.Lock()
+ w.lists[prefix] = l
+ w.mu.Unlock()
+ }
+ return keysOf(l.kes), l.err
+}
+
+func (w *walkStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
+ w.mu.Lock()
+ d, ok := w.dirs[prefix]
+ w.mu.Unlock()
+ if !ok {
+ dirs, err := w.Store.ListDirs(ctx, prefix)
+ d = dirListing{dirs, err}
+ w.mu.Lock()
+ w.dirs[prefix] = d
+ w.mu.Unlock()
+ }
+ return slices.Clone(d.dirs), d.err
+}
+
+func (w *walkStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ var prev *object
+ if o, ok := w.prev.lookup(key); ok && o.err == nil {
+ prev = &o
+ }
+ o := condGet(ctx, w.Store, key, prev)
+ w.mu.Lock()
+ w.gets[key] = o
+ w.mu.Unlock()
+ return o.result()
+}
+
+func keysOf(kes []release.KeyETag) []string {
+ if kes == nil {
+ return nil
+ }
+ out := make([]string, len(kes))
+ for i, ke := range kes {
+ out[i] = ke.Key
+ }
+ return out
+}
diff --git a/internal/provision/recordview_store.go b/internal/provision/recordview_store.go
new file mode 100644
index 000000000..939d87ad8
--- /dev/null
+++ b/internal/provision/recordview_store.go
@@ -0,0 +1,460 @@
+package provision
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "maps"
+ "os"
+ "slices"
+ "strings"
+ "sync"
+
+ "github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// ViewSnapshot is one immutable publication of a RecordView. Epoch increases
+// with every publication. Complete is false while any scope has a failed or
+// outstanding re-read, after a walk that could not list an account's apps,
+// or when the record walk over the snapshot would itself be partial (an
+// account manifest that will not read cleanly, an app listing that failed).
+// It is never read and changed at once: a Refresh builds a new one.
+type ViewSnapshot struct {
+ Epoch uint64
+ Complete bool
+
+ accounts map[string]*acctEntry // sqid -> accounts/<sqid>/
+ strays []string // keys under accounts/ outside every account scope, from the last walk
+ apps map[string]map[string]*appEntry // acct -> app -> apps/<acct>/<app>/
+ dirErrs map[string]error // acct -> why the last walk could not list apps/<acct>/
+}
+
+// acctEntry is one account scope: its complete key listing and account.json,
+// or the listing failure (which the manifest then carries too, so a reader
+// asking for the manifest learns the read failed rather than that it is
+// absent).
+type acctEntry struct {
+ keys []release.KeyETag
+ listErr error
+ manifest object
+}
+
+// appEntry is one app scope: both record listings (or why each failed), and
+// objs, every listed record body plus the two manifests and the trash marker.
+type appEntry struct {
+ inst, side []release.KeyETag
+ instErr, sideErr error
+ objs map[string]object
+}
+
+// object is one cached read: a body and its ETag, a not-found error wrapping
+// os.ErrNotExist, or the read error itself.
+type object struct {
+ body []byte
+ etag string
+ err error
+}
+
+func absent(key string) object {
+ return object{err: fmt.Errorf("provision: record view: %s: %w", key, os.ErrNotExist)}
+}
+
+// failed reports a read that did not answer: anything but a body or absence.
+func (o object) failed() bool { return o.err != nil && !errors.Is(o.err, os.ErrNotExist) }
+
+// result is the object as Get returns it, with a body the caller may keep.
+func (o object) result() ([]byte, string, error) {
+ if o.err != nil {
+ return nil, "", o.err
+ }
+ return slices.Clone(o.body), o.etag, nil
+}
+
+func accountManifestKey(a string) string { return "accounts/" + a + "/account.json" }
+
+func (e *acctEntry) empty() bool {
+ return e == nil || (len(e.keys) == 0 && e.listErr == nil)
+}
+
+// empty reports an app with no record-family state at all: nothing listed,
+// nothing failed, and all three per-app objects absent. A purged app is
+// dropped from the view, and from ListDirs over it.
+func (e *appEntry) empty() bool {
+ if e == nil {
+ return true
+ }
+ if len(e.inst) > 0 || len(e.side) > 0 || e.instErr != nil || e.sideErr != nil {
+ return false
+ }
+ for _, o := range e.objs {
+ if !errors.Is(o.err, os.ErrNotExist) {
+ return false
+ }
+ }
+ return true
+}
+
+// object is the cached read of key, or nil when the entry holds none.
+func (e *appEntry) object(key string) *object {
+ if e == nil {
+ return nil
+ }
+ o, ok := e.objs[key]
+ if !ok {
+ return nil
+ }
+ return &o
+}
+
+func (sn *ViewSnapshot) app(a, b string) *appEntry {
+ if sn == nil {
+ return nil
+ }
+ return sn.apps[a][b]
+}
+
+// clone copies the maps a Refresh changes; entries are shared, never mutated.
+func (sn *ViewSnapshot) clone() *ViewSnapshot {
+ out := &ViewSnapshot{
+ accounts: maps.Clone(sn.accounts),
+ strays: sn.strays,
+ apps: make(map[string]map[string]*appEntry, len(sn.apps)),
+ dirErrs: sn.dirErrs,
+ }
+ for a, inner := range sn.apps {
+ out.apps[a] = maps.Clone(inner)
+ }
+ return out
+}
+
+func (sn *ViewSnapshot) setAccount(a string, e *acctEntry) {
+ if e.empty() {
+ delete(sn.accounts, a)
+ return
+ }
+ sn.accounts[a] = e
+}
+
+func (sn *ViewSnapshot) setApp(a, b string, e *appEntry) {
+ if e.empty() {
+ delete(sn.apps[a], b)
+ if len(sn.apps[a]) == 0 {
+ delete(sn.apps, a)
+ }
+ return
+ }
+ if sn.apps[a] == nil {
+ sn.apps[a] = map[string]*appEntry{}
+ }
+ sn.apps[a][b] = e
+}
+
+// appKey splits apps/<a>/<b>/<rest> with all three parts non-empty.
+func appKey(key string) (a, b, rest string, ok bool) {
+ parts := strings.SplitN(key, "/", 4)
+ if len(parts) < 4 || parts[0] != "apps" || parts[1] == "" || parts[2] == "" || parts[3] == "" {
+ return "", "", "", false
+ }
+ return parts[1], parts[2], parts[3], true
+}
+
+// lookup answers Get of a key in a cached family; handled is false for every
+// other key. A key the view does not hold is absent: its family's listing is
+// complete. A record under a listing that failed answers that failure.
+func (sn *ViewSnapshot) lookup(key string) (o object, handled bool) {
+ if sn == nil {
+ return object{}, false
+ }
+ if a, ok := domains.AccountManifestSqid(key); ok {
+ e := sn.accounts[a]
+ if e == nil {
+ return absent(key), true
+ }
+ return e.manifest, true
+ }
+ a, b, rest, ok := appKey(key)
+ if !ok {
+ return object{}, false
+ }
+ var listErr error
+ switch {
+ case rest == "instances.json" || rest == "sidecars.json" || rest == "deleted.json":
+ case strings.HasPrefix(rest, "instances/") && rest != "instances/":
+ listErr = sn.app(a, b).listErr(true)
+ case strings.HasPrefix(rest, "sidecars/") && rest != "sidecars/":
+ listErr = sn.app(a, b).listErr(false)
+ default:
+ return object{}, false
+ }
+ if listErr != nil {
+ return object{err: listErr}, true
+ }
+ if p := sn.app(a, b).object(key); p != nil {
+ return *p, true
+ }
+ return absent(key), true
+}
+
+func (e *appEntry) listErr(inst bool) error {
+ switch {
+ case e == nil:
+ return nil
+ case inst:
+ return e.instErr
+ }
+ return e.sideErr
+}
+
+// list answers List of a cached family, sorted as the stores list.
+func (sn *ViewSnapshot) list(prefix string) (keys []string, handled bool, err error) {
+ if prefix == "accounts/" {
+ out := slices.Clone(sn.strays)
+ for a, e := range sn.accounts {
+ if e.listErr != nil {
+ // Unknown keys; the manifest is listed so a reader asks for it
+ // and learns the read failed (knownAccounts' readFailed).
+ out = append(out, accountManifestKey(a))
+ continue
+ }
+ out = append(out, keysOf(e.keys)...)
+ }
+ slices.Sort(out)
+ return out, true, nil
+ }
+ if a, ok := strings.CutPrefix(prefix, "accounts/"); ok {
+ a, ok = strings.CutSuffix(a, "/")
+ if !ok || a == "" || strings.Contains(a, "/") {
+ return nil, false, nil
+ }
+ e := sn.accounts[a]
+ if e == nil {
+ return nil, true, nil
+ }
+ return keysOf(e.keys), true, e.listErr
+ }
+ a, b, rest, ok := appKey(prefix)
+ if !ok || (rest != "instances/" && rest != "sidecars/") {
+ return nil, false, nil
+ }
+ e := sn.app(a, b)
+ if e == nil {
+ return nil, true, nil
+ }
+ if rest == "instances/" {
+ return keysOf(e.inst), true, e.instErr
+ }
+ return keysOf(e.side), true, e.sideErr
+}
+
+// listDirs answers ListDirs("apps/") and ListDirs("apps/<a>/"): accounts and
+// apps holding record-family state, as full prefixes, sorted. An account whose
+// apps the last walk could not list is listed, and listing it fails, as it did.
+func (sn *ViewSnapshot) listDirs(prefix string) (dirs []string, handled bool, err error) {
+ if prefix == "apps/" {
+ var out []string
+ for a := range sn.apps {
+ out = append(out, "apps/"+a+"/")
+ }
+ for a := range sn.dirErrs {
+ if sn.apps[a] == nil {
+ out = append(out, "apps/"+a+"/")
+ }
+ }
+ slices.Sort(out)
+ return out, true, nil
+ }
+ a, ok := strings.CutPrefix(prefix, "apps/")
+ if !ok {
+ return nil, false, nil
+ }
+ a, ok = strings.CutSuffix(a, "/")
+ if !ok || a == "" || strings.Contains(a, "/") {
+ return nil, false, nil
+ }
+ if err := sn.dirErrs[a]; err != nil {
+ return nil, true, err
+ }
+ var out []string
+ for b := range sn.apps[a] {
+ out = append(out, prefix+b+"/")
+ }
+ slices.Sort(out)
+ return out, true, nil
+}
+
+// Store is the snapshot as a release.Store over live: the cached families are
+// answered from the snapshot, everything else from live. Every write goes to
+// live, and from then on that key, and every listing it falls under, is read
+// from live for the rest of this store's life, so a caller reads its own
+// writes and a CAS retry re-reads the bucket. Each call returns a fresh store;
+// the snapshot itself never changes.
+func (sn *ViewSnapshot) Store(live release.Store) release.Store {
+ return &snapStore{sn: sn, live: live}
+}
+
+type snapStore struct {
+ sn *ViewSnapshot
+ live release.Store
+
+ mu sync.Mutex
+ wrote []string
+}
+
+var (
+ _ release.Store = (*snapStore)(nil)
+ _ release.BucketListDirs = (*snapStore)(nil)
+ _ release.Opener = (*snapStore)(nil)
+ _ release.PutStreamer = (*snapStore)(nil)
+ _ release.ConditionalDeleter = (*snapStore)(nil)
+ _ release.ModLister = (*snapStore)(nil)
+ _ release.SizeLister = (*snapStore)(nil)
+ _ release.PageLister = (*snapStore)(nil)
+ _ release.ConditionalGetter = (*snapStore)(nil)
+ _ release.ETagLister = (*snapStore)(nil)
+ _ release.AppMutationRunner = (*snapStore)(nil)
+)
+
+func (s *snapStore) mark(key string) {
+ s.mu.Lock()
+ s.wrote = append(s.wrote, key)
+ s.mu.Unlock()
+}
+
+// wroteUnder reports a write through this store to a key under prefix (or,
+// with exact, to key itself).
+func (s *snapStore) wroteUnder(prefix string, exact bool) bool {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ for _, k := range s.wrote {
+ if k == prefix || (!exact && strings.HasPrefix(k, prefix)) {
+ return true
+ }
+ }
+ return false
+}
+
+// BucketShapedListDirs: the snapshot answers ListDirs in the bucket's shape,
+// so the record walk lists per app instead of the whole of apps/.
+func (s *snapStore) BucketShapedListDirs() bool { return true }
+
+func (s *snapStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if !s.wroteUnder(key, true) {
+ if o, ok := s.sn.lookup(key); ok {
+ return o.result()
+ }
+ }
+ return s.live.Get(ctx, key)
+}
+
+func (s *snapStore) List(ctx context.Context, prefix string) ([]string, error) {
+ if !s.wroteUnder(prefix, false) {
+ if keys, ok, err := s.sn.list(prefix); ok {
+ return keys, err
+ }
+ }
+ return s.live.List(ctx, prefix)
+}
+
+func (s *snapStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
+ norm := prefix
+ if norm != "" && !strings.HasSuffix(norm, "/") {
+ norm += "/"
+ }
+ if !s.wroteUnder(norm, false) {
+ if dirs, ok, err := s.sn.listDirs(norm); ok {
+ return dirs, err
+ }
+ }
+ return s.live.ListDirs(ctx, prefix)
+}
+
+func (s *snapStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ s.mark(key)
+ return s.live.Put(ctx, key, body, ifMatch)
+}
+
+func (s *snapStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
+ s.mark(key)
+ return release.PutObjectFrom(ctx, s.live, key, body, size, ifMatch)
+}
+
+func (s *snapStore) Delete(ctx context.Context, key string) error {
+ s.mark(key)
+ return s.live.Delete(ctx, key)
+}
+
+func (s *snapStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
+ s.mark(key)
+ return release.DeleteIfMatch(ctx, s.live, key, etag)
+}
+
+// The remaining extensions are not cached families: forwarded to live, and
+// errors.ErrUnsupported when live lacks one.
+
+func (s *snapStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ return release.OpenObject(ctx, s.live, key)
+}
+
+func (s *snapStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
+ if g, ok := s.live.(release.ConditionalGetter); ok {
+ return g.GetIfChanged(ctx, key, etag)
+ }
+ return nil, "", false, errors.ErrUnsupported
+}
+
+func (s *snapStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
+ if l, ok := s.live.(release.ETagLister); ok {
+ return l.ListETags(ctx, prefix)
+ }
+ return nil, errors.ErrUnsupported
+}
+
+func (s *snapStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
+ if l, ok := s.live.(release.ModLister); ok {
+ return l.ListMod(ctx, prefix)
+ }
+ return nil, errors.ErrUnsupported
+}
+
+func (s *snapStore) ListSizes(ctx context.Context, prefix string) ([]release.KeySize, error) {
+ if l, ok := s.live.(release.SizeLister); ok {
+ return l.ListSizes(ctx, prefix)
+ }
+ return nil, errors.ErrUnsupported
+}
+
+func (s *snapStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
+ if l, ok := s.live.(release.PageLister); ok {
+ return l.ListPage(ctx, prefix, startAfter, limit)
+ }
+ return nil, errors.ErrUnsupported
+}
+
+// RunAppMutation keeps live's claim boundary for writes made through the
+// snapshot store; a live store without one runs f directly, as
+// release.WithAppMutation would.
+func (s *snapStore) RunAppMutation(ctx context.Context, acct, app, action string, f func(context.Context) error) error {
+ if r, ok := s.live.(release.AppMutationRunner); ok {
+ return r.RunAppMutation(ctx, acct, app, action, f)
+ }
+ return f(ctx)
+}
+
+// uncachedStore refuses everything. walkComplete runs the record walk over a
+// snapshot store backed by it, so a read the cache does not answer shows up
+// as a failure instead of a bucket request.
+type uncachedStore struct{}
+
+var errUncached = errors.New("provision: record view: not in the snapshot")
+
+func (uncachedStore) Get(context.Context, string) ([]byte, string, error) {
+ return nil, "", errUncached
+}
+func (uncachedStore) Put(context.Context, string, []byte, string) (string, error) {
+ return "", errUncached
+}
+func (uncachedStore) List(context.Context, string) ([]string, error) { return nil, errUncached }
+func (uncachedStore) ListDirs(context.Context, string) ([]string, error) { return nil, errUncached }
+func (uncachedStore) Delete(context.Context, string) error { return errUncached }
diff --git a/internal/provision/recordview_test.go b/internal/provision/recordview_test.go
new file mode 100644
index 000000000..adf6dc2be
--- /dev/null
+++ b/internal/provision/recordview_test.go
@@ -0,0 +1,798 @@
+package provision
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "reflect"
+ "slices"
+ "sort"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// viewCountingStore is a DirStore that counts every read by method and key,
+// and can fail every listing of one prefix. Embedding the concrete store
+// forwards every optional extension; bucket decides whether it promises
+// bucket-shaped ListDirs (the S3 store does, DirStore does not).
+type viewCountingStore struct {
+ *release.DirStore
+ bucket bool
+
+ mu sync.Mutex
+ calls []string
+ failList string // List, ListETags and ListDirs of exactly this prefix fail
+ failGet string // Get and GetIfChanged of exactly this key fail
+ ghost string // ListETags lists this key, which is not there: deleted between list and read
+ cancelOn string // ListETags of exactly this prefix calls cancel first
+ cancel context.CancelFunc
+}
+
+func (s *viewCountingStore) note(method, key string) {
+ s.mu.Lock()
+ s.calls = append(s.calls, method+" "+key)
+ s.mu.Unlock()
+}
+
+func (s *viewCountingStore) listFails(prefix string) bool {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.failList != "" && prefix == s.failList
+}
+
+func (s *viewCountingStore) getFails(key string) bool {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.failGet != "" && key == s.failGet
+}
+
+func (s *viewCountingStore) setFailList(p string) { s.mu.Lock(); s.failList = p; s.mu.Unlock() }
+func (s *viewCountingStore) setFailGet(k string) { s.mu.Lock(); s.failGet = k; s.mu.Unlock() }
+func (s *viewCountingStore) setGhost(k string) { s.mu.Lock(); s.ghost = k; s.mu.Unlock() }
+
+func (s *viewCountingStore) reset() {
+ s.mu.Lock()
+ s.calls = nil
+ s.mu.Unlock()
+}
+
+func (s *viewCountingStore) got() []string {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ out := slices.Clone(s.calls)
+ sort.Strings(out)
+ return out
+}
+
+func (s *viewCountingStore) BucketShapedListDirs() bool { return s.bucket }
+
+func (s *viewCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.note("Get", key)
+ if s.getFails(key) {
+ return nil, "", errors.New("get refused")
+ }
+ return s.DirStore.Get(ctx, key)
+}
+
+func (s *viewCountingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
+ s.note("GetIfChanged", key)
+ if s.getFails(key) {
+ return nil, "", false, errors.New("get refused")
+ }
+ return s.DirStore.GetIfChanged(ctx, key, etag)
+}
+
+func (s *viewCountingStore) List(ctx context.Context, prefix string) ([]string, error) {
+ s.note("List", prefix)
+ if s.listFails(prefix) {
+ return nil, errors.New("list refused")
+ }
+ return s.DirStore.List(ctx, prefix)
+}
+
+func (s *viewCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
+ s.note("ListETags", prefix)
+ s.mu.Lock()
+ if s.cancelOn != "" && prefix == s.cancelOn {
+ s.cancel()
+ }
+ s.mu.Unlock()
+ if s.listFails(prefix) {
+ return nil, errors.New("list refused")
+ }
+ kes, err := s.DirStore.ListETags(ctx, prefix)
+ s.mu.Lock()
+ ghost := s.ghost
+ s.mu.Unlock()
+ if err == nil && ghost != "" && strings.HasPrefix(ghost, prefix) {
+ kes = append(kes, release.KeyETag{Key: ghost, ETag: "ghost"})
+ slices.SortFunc(kes, func(x, y release.KeyETag) int { return strings.Compare(x.Key, y.Key) })
+ }
+ return kes, err
+}
+
+func (s *viewCountingStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
+ s.note("ListDirs", prefix)
+ if s.listFails(prefix) {
+ return nil, errors.New("listdirs refused")
+ }
+ return s.DirStore.ListDirs(ctx, prefix)
+}
+
+func (s *viewCountingStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
+ s.note("ListMod", prefix)
+ return s.DirStore.ListMod(ctx, prefix)
+}
+
+// viewFixture is a deployment of three accounts and five apps: records,
+// sidecars, both manifests and a trash marker. live writes bump the change
+// index; bypass writes the same directory as an old binary would, without
+// bumping it.
+type viewFixture struct {
+ live *viewCountingStore
+ bypass *release.DirStore
+ acct [3]string
+}
+
+func newViewFixture(t *testing.T, bucket, indexed bool) *viewFixture {
+ t.Helper()
+ root := t.TempDir()
+ d := release.NewDirStore(root)
+ d.EnableChangeIndex()
+ f := &viewFixture{live: &viewCountingStore{DirStore: d, bucket: bucket}, bypass: release.NewDirStore(root)}
+ var w release.Store = f.live
+ if !indexed {
+ w = f.bypass
+ }
+ for i := range f.acct {
+ f.acct[i] = writeAccountManifest(t, w, int64(301+i), fmt.Sprintf("Account %d", i))
+ }
+ a, b, c := f.acct[0], f.acct[1], f.acct[2]
+ for _, m := range []struct{ acct, app string }{{a, "shop"}, {a, "blog"}, {b, "docs"}, {c, "store"}} {
+ if err := PutManifest(ctx, w, m.acct, m.app, Manifest{Schema: 1, App: m.app, Domain: m.app + ".example.com"}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ f.record(t, w, a, "shop", "one")
+ f.record(t, w, a, "shop", "two")
+ f.record(t, w, a, "blog", "one")
+ f.record(t, w, b, "docs", "one")
+ f.record(t, w, c, "store", "one")
+ // A record that reads but does not decode: unreadable, by its path host.
+ if _, err := w.Put(ctx, RecordKey(c, "store", "bad.store.example.com"), []byte("{not json"), ""); err != nil {
+ t.Fatal(err)
+ }
+ for _, m := range []struct{ acct, app string }{{a, "shop"}, {b, "wiki"}} {
+ if err := PutSidecarManifest(ctx, w, m.acct, m.app, SidecarManifest{Schema: 1, App: m.app, Args: []string{"run"}, Channel: "stable"}); err != nil {
+ t.Fatal(err)
+ }
+ f.sidecar(t, w, m.acct, m.app, "bot")
+ }
+ f.trash(t, w, a, "blog")
+ // Keys outside the record families.
+ for _, k := range []string{
+ "apps/" + a + "/shop/logs/2026/10/21/chunk-1.jsonl",
+ "apps/" + a + "/shop/channels/stable.json",
+ "apps/" + a + "/shop/config.json",
+ "accounts/" + a + "/members/owner.json",
+ } {
+ if _, err := w.Put(ctx, k, []byte(`{}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ f.live.reset()
+ return f
+}
+
+func viewRecord(acct, app, slug string) Record {
+ return Record{Schema: 1, Host: slug + "." + app + ".example.com", Slug: slug, App: app,
+ Account: acct, Region: "ie", Channel: "stable", CreatedAt: time.Now()}
+}
+
+func (f *viewFixture) record(t *testing.T, s release.Store, acct, app, slug string) {
+ t.Helper()
+ if err := CreateRecord(ctx, s, viewRecord(acct, app, slug)); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (f *viewFixture) sidecar(t *testing.T, s release.Store, acct, app, name string) {
+ t.Helper()
+ rec := SidecarRecord{Schema: 1, Name: name, App: app, Account: acct, Host: SidecarHost(name, app, acct), Channel: "stable"}
+ if err := CreateSidecarRecord(ctx, s, rec); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (f *viewFixture) trash(t *testing.T, s release.Store, acct, app string) {
+ t.Helper()
+ body := mustJSON(t, release.AppTombstone{Schema: 1, Account: acct, App: app, DeletedAt: time.Now().UTC(), Actor: "t"})
+ if _, err := s.Put(ctx, release.AppTombstoneKey(acct, app), body, ""); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (f *viewFixture) view() *RecordView {
+ return NewRecordView(f.live, release.NewChangeWatch(f.live))
+}
+
+func mustRebuild(t *testing.T, v *RecordView) *ViewSnapshot {
+ t.Helper()
+ if err := v.Rebuild(ctx); err != nil {
+ t.Fatal(err)
+ }
+ sn := v.Snapshot()
+ if sn == nil {
+ t.Fatal("Rebuild published nothing")
+ }
+ return sn
+}
+
+func mustRefresh(t *testing.T, v *RecordView) *ViewSnapshot {
+ t.Helper()
+ if err := v.Refresh(ctx); err != nil {
+ t.Fatal(err)
+ }
+ sn := v.Snapshot()
+ if sn == nil {
+ t.Fatal("Refresh published nothing")
+ }
+ return sn
+}
+
+// walkAnswer is everything the record consumers read, as one comparable
+// value: the strict record set and its unreadable hosts, the known accounts,
+// each app's sidecar records, both manifests and the trash marker.
+type walkAnswer struct {
+ Records []Record
+ Unreadable map[string]bool
+ Known map[string]bool
+ ReadFailed bool
+ PerApp map[string]string
+}
+
+func readWalk(t *testing.T, s release.Store, f *viewFixture) walkAnswer {
+ t.Helper()
+ recs, unreadable, err := ListAllRecordsStrict(ctx, s)
+ if err != nil {
+ t.Fatalf("ListAllRecordsStrict: %v", err)
+ }
+ for i := range recs {
+ recs[i].CreatedAt = recs[i].CreatedAt.UTC()
+ }
+ known, readFailed, err := knownAccounts(ctx, s)
+ if err != nil {
+ t.Fatalf("knownAccounts: %v", err)
+ }
+ per := map[string]string{}
+ for _, acct := range f.acct {
+ for _, app := range []string{"shop", "blog", "docs", "wiki", "store", "fresh"} {
+ m, merr := GetManifest(ctx, s, acct, app)
+ sm, serr := GetSidecarManifest(ctx, s, acct, app)
+ tomb, terr := release.ReadAppTombstone(ctx, s, acct, app)
+ sides, _, lerr := ListSidecarRecords(ctx, s, acct, app, 100)
+ per[acct+"/"+app] = fmt.Sprintf("m=%s/%v sm=%s/%v tomb=%v/%v sides=%s/%v",
+ mustJSON(t, m), errClass(merr), mustJSON(t, sm), errClass(serr), tomb != nil, errClass(terr), mustJSON(t, sides), errClass(lerr))
+ }
+ }
+ return walkAnswer{Records: recs, Unreadable: unreadable, Known: known, ReadFailed: readFailed, PerApp: per}
+}
+
+func errClass(err error) string {
+ switch {
+ case err == nil:
+ return "ok"
+ case errors.Is(err, os.ErrNotExist):
+ return "absent"
+ }
+ return "error: " + err.Error()
+}
+
+func assertMatchesWalk(t *testing.T, f *viewFixture, sn *ViewSnapshot) {
+ t.Helper()
+ want := readWalk(t, f.live, f)
+ got := readWalk(t, sn.Store(f.live), f)
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("over the snapshot:\n%+v\nover the bucket:\n%+v", got, want)
+ }
+}
+
+func viewHosts(t *testing.T, s release.Store) []string {
+ t.Helper()
+ recs, err := ListAllRecords(ctx, s)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var hosts []string
+ for _, r := range recs {
+ hosts = append(hosts, r.Host)
+ }
+ return hosts
+}
+
+func TestViewMatchesWalk(t *testing.T) {
+ t.Parallel()
+ for _, bucket := range []bool{false, true} {
+ t.Run(fmt.Sprintf("bucket=%v", bucket), func(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, bucket, true)
+ v := f.view()
+ sn := mustRefresh(t, v)
+ if !sn.Complete {
+ t.Fatal("a clean first build is not complete")
+ }
+ if !release.HasBucketListDirs(sn.Store(f.live)) {
+ t.Fatal("the snapshot store does not promise bucket-shaped ListDirs, so the record walk would list apps/ whole")
+ }
+ assertMatchesWalk(t, f, sn)
+ if len(readWalk(t, f.live, f).Records) != 5 {
+ t.Fatal("fixture: want five readable records")
+ }
+
+ // Incremental changes in every family, then a Refresh.
+ a, b, c := f.acct[0], f.acct[1], f.acct[2]
+ f.record(t, f.live, b, "docs", "two")
+ moved := viewRecord(a, "shop", "one")
+ moved.Region = "us"
+ putReplace(t, f.live, RecordKey(a, "shop", moved.Host), mustJSON(t, moved))
+ f.record(t, f.live, a, "fresh", "one")
+ f.trash(t, f.live, a, "shop")
+ if err := f.live.Delete(ctx, release.AppTombstoneKey(a, "blog")); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.live.Delete(ctx, SidecarRecordKey(b, "wiki", "bot")); err != nil {
+ t.Fatal(err)
+ }
+ if err := PutManifest(ctx, f.live, c, "store", Manifest{Schema: 1, App: "store", Domain: "other.example.com"}); err != nil {
+ t.Fatal(err)
+ }
+ soft := domains.AccountManifest{ID: 303, Sqid: c, Name: "gone", CreatedAt: time.Now().UTC(), DeletedAt: time.Now().UTC()}
+ putReplace(t, f.live, "accounts/"+c+"/account.json", mustJSON(t, soft))
+ sn2 := mustRefresh(t, v)
+ if sn2.Epoch <= sn.Epoch || !sn2.Complete {
+ t.Fatalf("refresh published epoch %d complete %v after %d", sn2.Epoch, sn2.Complete, sn.Epoch)
+ }
+ assertMatchesWalk(t, f, sn2)
+ })
+ }
+}
+
+// Complete is false exactly when the walk over the bucket would be partial:
+// an account whose apps will not list, or an account manifest that will not
+// read cleanly. A strict reader over the snapshot fails as it does live.
+func TestViewCompleteMirrorsTheWalk(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ a, b := f.acct[0], f.acct[1]
+ v := f.view()
+ f.live.setFailList("apps/" + b + "/")
+ f.live.setGhost(RecordKey(a, "shop", "ghost.shop.example.com"))
+ sn := mustRebuild(t, v)
+ if sn.Complete {
+ t.Fatal("a walk that could not list an account's apps is complete")
+ }
+ if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
+ t.Fatal("a strict reader over a partial walk did not fail")
+ }
+ f.live.setFailList("")
+ sn = mustRefresh(t, v)
+ if !sn.Complete {
+ t.Fatal("the refresh after a partial walk did not walk again")
+ }
+ assertMatchesWalk(t, f, sn)
+
+ putReplace(t, f.live, "accounts/"+b+"/account.json", []byte("{not json"))
+ sn = mustRefresh(t, v)
+ if sn.Complete {
+ t.Fatal("a snapshot holding a malformed account manifest is complete")
+ }
+ _, _, lerr := ListAllRecordsStrict(ctx, f.live)
+ _, _, serr := ListAllRecordsStrict(ctx, sn.Store(f.live))
+ if lerr == nil || serr == nil {
+ t.Fatalf("strict readers over a malformed manifest: live %v, snapshot %v; want both to fail", lerr, serr)
+ }
+
+ // An account listing that fails: its manifest reads as failed, not absent.
+ f.live.setFailList("accounts/" + b + "/")
+ putReplace(t, f.live, "accounts/"+b+"/account.json", accountManifestBody(302, b, "Account 1"))
+ sn = mustRefresh(t, v)
+ if sn.Complete {
+ t.Fatal("a snapshot whose account listing failed is complete")
+ }
+ if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
+ t.Fatal("a strict reader over a failed account listing did not fail")
+ }
+ f.live.setFailList("")
+ if sn = mustRefresh(t, v); !sn.Complete {
+ t.Fatal("still incomplete after the account listing recovered")
+ }
+ assertMatchesWalk(t, f, sn)
+}
+
+// A refresh whose caller goes away mid-read publishes nothing: reads that
+// died with the caller are not the bucket's answer.
+func TestViewCancelledRefreshPublishesNothing(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ before := mustRebuild(t, v)
+ a := f.acct[0]
+ f.record(t, f.live.DirStore, a, "shop", "three")
+ cctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+ f.live.mu.Lock()
+ f.live.cancelOn, f.live.cancel = RecordsPrefix(a, "shop"), cancel
+ f.live.mu.Unlock()
+ if err := v.Refresh(cctx); err == nil {
+ t.Fatal("a refresh cancelled mid-read reported success")
+ }
+ if v.Snapshot() != before {
+ t.Fatal("a refresh cancelled mid-read published")
+ }
+ f.live.mu.Lock()
+ f.live.cancelOn = ""
+ f.live.mu.Unlock()
+ if hosts := viewHosts(t, mustRefresh(t, v).Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
+ t.Fatalf("records after the next refresh = %v", hosts)
+ }
+}
+
+func TestViewRefreshRereadsOnlyMovedApp(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ mustRebuild(t, v)
+ a := f.acct[0]
+ f.record(t, f.live.DirStore, a, "shop", "three")
+ f.live.reset()
+ sn := mustRefresh(t, v)
+ app := "apps/" + a + "/shop/"
+ want := []string{
+ "GetIfChanged " + app + "deleted.json",
+ "GetIfChanged " + app + "instances.json",
+ "GetIfChanged " + app + "instances/three.shop.example.com.json",
+ "GetIfChanged " + app + "sidecars.json",
+ "GetIfChanged " + release.ChangeIndexKey,
+ "ListETags " + app + "instances/",
+ "ListETags " + app + "sidecars/",
+ }
+ if got := f.live.got(); !reflect.DeepEqual(got, want) {
+ t.Fatalf("one record in shop read\n%s\nwant\n%s", strings.Join(got, "\n"), strings.Join(want, "\n"))
+ }
+ if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
+ t.Fatalf("records over the snapshot = %v, want the new one", hosts)
+ }
+}
+
+func TestViewRefreshNoChangeIsOneGet(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ mustRebuild(t, v)
+ f.live.reset()
+ mustRefresh(t, v)
+ if got, want := f.live.got(), []string{"GetIfChanged " + release.ChangeIndexKey}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("a quiet refresh read %v, want %v", got, want)
+ }
+}
+
+func TestViewRetriesFailedScopeOn304(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ mustRebuild(t, v)
+ a := f.acct[0]
+ f.record(t, f.live.DirStore, a, "shop", "three")
+ f.live.setFailList(RecordsPrefix(a, "shop"))
+ sn := mustRefresh(t, v)
+ if sn.Complete {
+ t.Fatal("a scope whose re-read failed is reported complete")
+ }
+ if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
+ t.Fatal("a strict reader over an incomplete snapshot did not fail")
+ }
+ if _, complete, err := ListAllRecordsTolerant(ctx, sn.Store(f.live)); err != nil || complete {
+ t.Fatalf("a tolerant reader over an incomplete snapshot = complete %v, %v", complete, err)
+ }
+
+ f.live.setFailList("")
+ f.live.reset()
+ sn = mustRefresh(t, v)
+ if !slices.Contains(f.live.got(), "ListETags "+RecordsPrefix(a, "shop")) {
+ t.Fatalf("the failed scope was not re-read on an unchanged index: %v", f.live.got())
+ }
+ if slices.Contains(f.live.got(), "ListETags accounts/") || slices.Contains(f.live.got(), "List accounts/") {
+ t.Fatalf("the retry walked the deployment: %v", f.live.got())
+ }
+ if !sn.Complete {
+ t.Fatal("still incomplete after the retry succeeded")
+ }
+ if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
+ t.Fatalf("records after the retry = %v", hosts)
+ }
+}
+
+func TestViewFirstRefreshWithoutIndexRebuilds(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, false)
+ if _, _, err := f.bypass.Get(ctx, release.ChangeIndexKey); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("fixture wrote an index: %v", err)
+ }
+ v := f.view()
+ sn := mustRefresh(t, v)
+ if !sn.Complete {
+ t.Fatal("a first build over a bucket with no index is not complete")
+ }
+ assertMatchesWalk(t, f, sn)
+ if len(viewHosts(t, sn.Store(f.live))) != 5 {
+ t.Fatal("the first build missed records")
+ }
+}
+
+func TestViewUnusableWatchRebuilds(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ mustRebuild(t, v)
+ putReplace(t, f.bypass, release.ChangeIndexKey, []byte("{not json"))
+ f.record(t, f.bypass, f.acct[1], "docs", "two")
+ f.live.reset()
+ sn := mustRefresh(t, v)
+ if !slices.Contains(f.live.got(), "ListETags accounts/") {
+ t.Fatalf("an unusable index did not walk: %v", f.live.got())
+ }
+ if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "two.docs.example.com") {
+ t.Fatalf("records after the walk = %v", hosts)
+ }
+}
+
+func TestViewBypassWriteAppearsAfterRebuild(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ mustRebuild(t, v)
+ f.record(t, f.bypass, f.acct[0], "shop", "three")
+ sn := mustRefresh(t, v)
+ if hosts := viewHosts(t, sn.Store(f.live)); slices.Contains(hosts, "three.shop.example.com") {
+ t.Fatal("a write that bypassed the index appeared without a walk")
+ }
+ sn = mustRebuild(t, v)
+ if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
+ t.Fatalf("records after the backstop walk = %v", hosts)
+ }
+ assertMatchesWalk(t, f, sn)
+}
+
+func TestViewPurgeDropsApp(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ sn := mustRebuild(t, v)
+ a := f.acct[0]
+ dirs, err := sn.Store(f.live).ListDirs(ctx, "apps/"+a+"/")
+ if err != nil || !slices.Contains(dirs, "apps/"+a+"/shop/") {
+ t.Fatalf("ListDirs before the purge = %v, %v", dirs, err)
+ }
+ keys, err := f.live.DirStore.List(ctx, "apps/"+a+"/shop/")
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, k := range keys {
+ if err := f.live.DirStore.Delete(ctx, k); err != nil {
+ t.Fatal(err)
+ }
+ }
+ sn = mustRefresh(t, v)
+ dirs, err = sn.Store(f.live).ListDirs(ctx, "apps/"+a+"/")
+ if err != nil || slices.Contains(dirs, "apps/"+a+"/shop/") || !slices.Contains(dirs, "apps/"+a+"/blog/") {
+ t.Fatalf("ListDirs after the purge = %v, %v; want blog and no shop", dirs, err)
+ }
+ assertMatchesWalk(t, f, sn)
+}
+
+func TestViewEpochsIncrease(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ if v.Snapshot() != nil {
+ t.Fatal("a view that never published has a snapshot")
+ }
+ e1 := mustRebuild(t, v)
+ f.record(t, f.live.DirStore, f.acct[0], "shop", "three")
+ e2 := mustRefresh(t, v)
+ if e2.Epoch <= e1.Epoch {
+ t.Fatalf("refresh epoch %d after %d", e2.Epoch, e1.Epoch)
+ }
+ if again := mustRefresh(t, v); again != e2 {
+ t.Fatal("a refresh that read nothing published a new snapshot")
+ }
+ e3 := mustRebuild(t, v)
+ if e3.Epoch <= e2.Epoch {
+ t.Fatalf("rebuild epoch %d after %d", e3.Epoch, e2.Epoch)
+ }
+}
+
+// The backstop holds mu for its whole walk, so an incremental refresh that
+// starts during it waits, and runs after the rebuild publishes: the rebuild
+// cannot publish over what the refresh read. Serial: it sets a package hook.
+func TestViewBackstopCannotOverwriteNewer(t *testing.T) {
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ first := mustRebuild(t, v)
+
+ paused, resume := make(chan struct{}), make(chan struct{})
+ var once sync.Once
+ recordViewRebuildHook = func() {
+ once.Do(func() {
+ close(paused)
+ <-resume
+ })
+ }
+ t.Cleanup(func() { recordViewRebuildHook = nil })
+
+ rebuilt := make(chan error, 1)
+ go func() { rebuilt <- v.Rebuild(ctx) }()
+ <-paused
+ f.record(t, f.live.DirStore, f.acct[0], "shop", "three")
+ refreshed := make(chan error, 1)
+ go func() { refreshed <- v.Refresh(ctx) }()
+ select {
+ case err := <-refreshed:
+ t.Fatalf("a refresh ran while the backstop walk held the view: %v", err)
+ case <-time.After(100 * time.Millisecond):
+ }
+ close(resume)
+ if err := <-rebuilt; err != nil {
+ t.Fatal(err)
+ }
+ if err := <-refreshed; err != nil {
+ t.Fatal(err)
+ }
+ sn := v.Snapshot()
+ if sn.Epoch != first.Epoch+2 || !sn.Complete {
+ t.Fatalf("final epoch %d complete %v, want %d and complete", sn.Epoch, sn.Complete, first.Epoch+2)
+ }
+ if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
+ t.Fatalf("the record written during the walk is missing: %v", hosts)
+ }
+}
+
+func TestSnapshotStoreReadYourWrites(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ sn := mustRebuild(t, v)
+ a := f.acct[0]
+ st := sn.Store(f.live)
+ rec := viewRecord(a, "fresh", "one")
+ if err := CreateRecord(ctx, st, rec); err != nil {
+ t.Fatal(err)
+ }
+ if got, err := GetRecord(ctx, st, a, "fresh", rec.Host); err != nil || got.Host != rec.Host {
+ t.Fatalf("Get after Put = %+v, %v", got, err)
+ }
+ if keys, err := st.List(ctx, RecordsPrefix(a, "fresh")); err != nil || !slices.Contains(keys, RecordKey(a, "fresh", rec.Host)) {
+ t.Fatalf("List after Put = %v, %v", keys, err)
+ }
+ if dirs, err := st.ListDirs(ctx, "apps/"+a+"/"); err != nil || !slices.Contains(dirs, "apps/"+a+"/fresh/") {
+ t.Fatalf("ListDirs after Put = %v, %v", dirs, err)
+ }
+ if err := PutManifest(ctx, st, a, "shop", Manifest{Schema: 1, App: "shop", Domain: "moved.example.com"}); err != nil {
+ t.Fatal(err)
+ }
+ if m, err := GetManifest(ctx, st, a, "shop"); err != nil || m.Domain != "moved.example.com" {
+ t.Fatalf("manifest after a CAS write = %+v, %v", m, err)
+ }
+ gone := RecordKey(a, "shop", "one.shop.example.com")
+ if err := st.Delete(ctx, gone); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := st.Get(ctx, gone); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("Get after Delete = %v", err)
+ }
+ if keys, _ := st.List(ctx, RecordsPrefix(a, "shop")); slices.Contains(keys, gone) {
+ t.Fatalf("List after Delete still has %s", gone)
+ }
+ // The snapshot itself never moves: a fresh store over it still answers
+ // from the cache.
+ if keys, _ := sn.Store(f.live).List(ctx, RecordsPrefix(a, "shop")); !slices.Contains(keys, gone) {
+ t.Fatal("a write through one snapshot store changed the snapshot")
+ }
+}
+
+// Seeded without the index, so the failed scope is not in it: only the
+// view's own record of the failure brings it back on the next Refresh.
+func TestSnapshotStoreErrorsMatchLive(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, false)
+ a, b := f.acct[0], f.acct[1]
+ f.live.setFailGet(ManifestKey(b, "docs"))
+ v := f.view()
+ sn := mustRebuild(t, v)
+ if sn.Complete {
+ t.Fatal("a snapshot holding a failed manifest read is complete")
+ }
+ st := sn.Store(f.live)
+ for _, key := range []string{
+ RecordKey(a, "shop", "missing.shop.example.com"),
+ RecordKey(a, "nowhere", "missing.nowhere.example.com"),
+ SidecarRecordKey(a, "shop", "missing"),
+ ManifestKey(b, "wiki"),
+ SidecarManifestKey(a, "blog"),
+ release.AppTombstoneKey(a, "shop"),
+ "accounts/zzzz/account.json",
+ } {
+ _, _, lerr := f.live.DirStore.Get(ctx, key)
+ _, _, serr := st.Get(ctx, key)
+ if !errors.Is(lerr, os.ErrNotExist) || !errors.Is(serr, os.ErrNotExist) {
+ t.Fatalf("Get %s: live %v, snapshot %v; want not-found from both", key, lerr, serr)
+ }
+ }
+ if _, err := GetManifest(ctx, st, b, "docs"); err == nil || errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("a manifest whose read failed = %v, want the read error, not absence", err)
+ }
+ if tomb, err := release.ReadAppTombstone(ctx, st, a, "shop"); tomb != nil || err != nil {
+ t.Fatalf("absent trash marker = %+v, %v", tomb, err)
+ }
+ if tomb, err := release.ReadAppTombstone(ctx, st, a, "blog"); tomb == nil || err != nil {
+ t.Fatalf("trash marker = %+v, %v", tomb, err)
+ }
+
+ f.live.setFailGet("")
+ sn = mustRefresh(t, v)
+ if !sn.Complete {
+ t.Fatal("still incomplete after the manifest read recovered")
+ }
+ if m, err := GetManifest(ctx, sn.Store(f.live), b, "docs"); err != nil || m.App != "docs" {
+ t.Fatalf("recovered manifest = %+v, %v", m, err)
+ }
+}
+
+func TestSnapshotStoreOtherKeysGoLive(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ sn := mustRebuild(t, v)
+ a := f.acct[0]
+ if _, err := f.bypass.Put(ctx, "apps/"+a+"/shop/channels/edge.json", []byte(`{}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ putReplace(t, f.bypass, "apps/"+a+"/shop/config.json", []byte(`{"after":true}`))
+ f.live.reset()
+ st := sn.Store(f.live)
+ body, _, err := st.Get(ctx, "apps/"+a+"/shop/config.json")
+ if err != nil || string(body) != `{"after":true}` {
+ t.Fatalf("config.json over the snapshot = %s, %v", body, err)
+ }
+ keys, err := st.List(ctx, "apps/"+a+"/shop/channels/")
+ if err != nil || !slices.Contains(keys, "apps/"+a+"/shop/channels/edge.json") {
+ t.Fatalf("channels over the snapshot = %v, %v", keys, err)
+ }
+ want := []string{"Get apps/" + a + "/shop/config.json", "List apps/" + a + "/shop/channels/"}
+ if got := f.live.got(); !reflect.DeepEqual(got, want) {
+ t.Fatalf("live saw %v, want %v", got, want)
+ }
+}
+
+func TestViewScopeApp(t *testing.T) {
+ t.Parallel()
+ for scope, want := range map[string][2]string{
+ "apps/abc/shop": {"abc", "shop"},
+ "apps/abc": {},
+ "apps/abc/": {},
+ "apps//shop": {},
+ "apps/a/b/c": {},
+ "accounts/abc": {},
+ } {
+ acct, app, ok := ScopeApp(scope)
+ if ok != (want[0] != "") || acct != want[0] || app != want[1] {
+ t.Fatalf("ScopeApp(%q) = %q, %q, %v", scope, acct, app, ok)
+ }
+ }
+}