Documentation
¶
Overview ¶
Change feed for the work index (tasks/156): a best-effort accelerator that gives cross-container read-your-writes without a corpus List. A write appends its projected entry (or a tombstone) to a small feed blob; a reader replays the feed over its snapshot. The List-diff refresh (workindex.go) stays the correctness backstop, so every feed operation here is best-effort -- a failure is logged by the caller and ignored, never fatal.
The feed and snapshot share a fold epoch. When the feed would grow past the fold threshold (e.g. a bulk update), the writer folds instead: it saves a fresh snapshot at epoch+1 and resets the feed to empty, so the feed stays bounded and readers reload the new base rather than replaying a huge delta.
Snapshot persistence for the work index (tasks/155): a cold start loads the projection from one blob instead of GETting every grain. The snapshot is the in-memory projection (identity + merges + barcodes + summaries per grain), serialized as gzipped JSON -- portable, inspectable, and readable by the Rust/WASM side, unlike a Go-only gob stream. It is a disposable cache: a missing, corrupt, or wrong-version snapshot costs a full scan on the next boot, never correctness, because refreshLocked's ETag diff always reconciles.
Package workindex maintains an in-memory identity index over the work-grain tree: provider keys, clustering keys, barcodes, and work summaries per grain. It exists so interactive request paths cost O(1) blob reads instead of re-walking the corpus (tasks/106). Freshness is two-layered: reads refresh by ETag diff on a short TTL (one List per window; only changed grains are re-fetched and re-scanned), and the API's own write paths push their writes in synchronously via Apply, so a session always reads its own writes. Writers outside the process (or outside httpapi, until tasks/107 and 109 route them here) become visible within one TTL.
Index ¶
- Constants
- type Index
- func (ix *Index) AppendFeed(ctx context.Context, paths ...string) error
- func (ix *Index) Apply(grainPath, etag string, grain []byte)
- func (ix *Index) Barcodes(ctx context.Context) (map[string]bool, error)
- func (ix *Index) ClusterOwners(ctx context.Context, key string) ([]Ref, error)
- func (ix *Index) DuplicateGroups(ctx context.Context) (map[string][]string, error)
- func (ix *Index) GrainPaths(ctx context.Context) (map[string]bool, error)
- func (ix *Index) LoadSnapshot(ctx context.Context) error
- func (ix *Index) ProviderOwners(ctx context.Context, key string) ([]Ref, error)
- func (ix *Index) Refresh(ctx context.Context) error
- func (ix *Index) RefreshNow(ctx context.Context) error
- func (ix *Index) Save(ctx context.Context) error
- func (ix *Index) SeedResolver(ctx context.Context, r *identity.Resolver) error
- func (ix *Index) SetSnapshotPath(p string)
- func (ix *Index) SnapshotDrift() (primed, refetched int)
- func (ix *Index) Summaries(ctx context.Context) ([]ingest.WorkSummary, error)
- func (ix *Index) SummariesWithPaths(ctx context.Context) ([]ingest.WorkSummary, map[string]string, error)
- func (ix *Index) Update(ctx context.Context, paths ...string) error
- func (ix *Index) WarmScan(ctx context.Context, workers int, progress func(fetched, total int)) error
- type Ref
Constants ¶
const DefaultFeedPath = "data/workindex.feed"
DefaultFeedPath is where the change feed lives -- outside the grain prefix the index lists, like the snapshot.
const DefaultFeedTTL = 2 * time.Second
DefaultFeedTTL bounds how often a read re-polls the feed; within it, reads are served from memory.
const DefaultFoldThreshold = 4096
DefaultFoldThreshold is the feed record count past which an append folds into a fresh snapshot instead of growing the feed.
const DefaultSnapshotPath = "data/workindex.snapshot"
DefaultSnapshotPath is where the persisted projection lives -- outside the grain prefix the index lists, so refreshLocked never mistakes it for a grain.
const DefaultTTL = 30 * time.Second
DefaultTTL matches the editing UI's freshness contract established by the works-list cache: fresh after a publish, not per keystroke.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Index ¶
type Index struct {
// contains filtered or unexported fields
}
Index is the shared corpus index. All methods are safe for concurrent use; reads that find the index stale pay the refresh inline (ETag-diff List, so an unchanged corpus costs zero Gets).
func New ¶
New returns an index over the grains under prefix (normally "data/works/"). The first read pays the full corpus scan; subsequent reads are cache hits.
func (*Index) AppendFeed ¶
AppendFeed records the current state of the given grain paths (or a tombstone for a path no longer present) to the change feed, so other containers see the write without a corpus List. Call it after Apply/Update has committed the write in memory. Best-effort: callers log the error but do not fail the write, since the List-diff refresh is the backstop.
func (*Index) Apply ¶
Apply records a grain the caller just wrote (or re-read), keeping the index exact for the process's own writes without waiting out the TTL.
func (*Index) Barcodes ¶
Barcodes returns every barcode in the corpus. The map is a copy the caller may mutate (the bulk-add generator reserves candidates in it).
func (*Index) ClusterOwners ¶
ClusterOwners returns the works carrying the author/title/language clustering key, ordered by grain path.
func (*Index) DuplicateGroups ¶
DuplicateGroups returns cluster keys shared by two or more distinct works: key -> sorted work ids. The map is the caller's to keep.
func (*Index) GrainPaths ¶
GrainPaths returns the set of grain paths the index currently covers. The map is a copy the caller may keep.
func (*Index) LoadSnapshot ¶
LoadSnapshot primes the grain entries from the snapshot blob so a cold start skips the corpus scan. It leaves the refresh clock unset, so the next read still runs the ETag-diff reconcile -- re-reading only grains changed since the snapshot and dropping any deleted since. A missing snapshot is not an error (first boot). A corrupt or wrong-version one returns an error so the caller can log and fall back to the full scan.
func (*Index) ProviderOwners ¶
ProviderOwners returns the works whose instances carry the provider key (isbn:/issn:/id:-namespaced identifier), ordered by grain path.
func (*Index) Refresh ¶
Refresh makes the index fresh now if its TTL has lapsed -- the boot-time warmer's entry point; reads call it implicitly.
func (*Index) RefreshNow ¶
RefreshNow reconciles against a fresh listing regardless of the TTL -- for callers about to make corpus-accuracy-sensitive decisions (a copycat commit's re-match, tasks/107). Still an ETag diff: only changed grains are re-read.
func (*Index) Save ¶
Save serializes the current projection to the snapshot blob (gzipped JSON). It reads straight from memory -- no grain reads -- so it is cheap to call after a warm-up or a publish. Entries are written in path order so the artifact is deterministic.
func (*Index) SeedResolver ¶
SeedResolver seeds r with every committed work/instance identity and merge marker in the corpus, in grain path order -- what copycat's match pass needs from LoadPriorStore, without the per-request corpus read (tasks/107).
func (*Index) SetSnapshotPath ¶
SetSnapshotPath overrides where Save/LoadSnapshot read and write the persisted projection (default DefaultSnapshotPath). Call it before first use; the offline seed tool uses it to honor an --out flag.
func (*Index) SnapshotDrift ¶
SnapshotDrift reports how the first reconcile after a snapshot load treated the primed entries: primed is the entry count the snapshot supplied, refetched how many of those the ETag diff re-read anyway. refetched near primed means the snapshot was built against a store with a different ETag scheme (a dir-built seed copied into S3, tasks/162) -- correctness holds, but the boot degrades to the full corpus scan the snapshot was meant to avoid. Both are zero until a load-then-reconcile cycle completes.
func (*Index) Summaries ¶
Summaries returns every work's summary, sorted by work id (the same shape and order as ingest.ScanSummaries). The slice is shared: read-only.
func (*Index) SummariesWithPaths ¶
func (ix *Index) SummariesWithPaths(ctx context.Context) ([]ingest.WorkSummary, map[string]string, error)
SummariesWithPaths returns every work's summary plus each work's grain path -- the ingest.SummarySource contract the worker paths consume (tasks/109). Both return values are shared: read-only.
func (*Index) Update ¶
Update re-reads the given grain paths and applies their current state -- how a bulk writer that lands many grains at once (copycat commit/revert) keeps the index exact without waiting out the TTL. A path that no longer exists is dropped from the index.
func (*Index) WarmScan ¶
func (ix *Index) WarmScan(ctx context.Context, workers int, progress func(fetched, total int)) error
WarmScan reconciles against a fresh listing like RefreshNow, but fetches changed grains with workers concurrent GETs and reports progress after each -- the offline seed tool's path (tasks/162), where a sequential reconcile over a high-latency store takes hours. Not for concurrent use with writers: a grain written between the List and the final merge may be recorded stale until the next refresh. progress may be nil.
Source Files