Documentation
¶
Overview ¶
Change feed for the work index: 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: 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. 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, and 109 route them here) become visible within one TTL.
Index ¶
- Constants
- type BarcodeHolder
- type DuplicateBarcode
- type Index
- func (ix *Index) AllocateBarcodes(fn func() error) error
- func (ix *Index) AppendFeed(ctx context.Context, paths ...string) error
- func (ix *Index) Apply(grainPath, etag string, grain []byte)
- func (ix *Index) BarcodeHeldByOther(ctx context.Context, barcode, workID, instanceID string) (bool, BarcodeHolder, error)
- 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) DuplicateBarcodes(ctx context.Context) ([]DuplicateBarcode, 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) MergedInto(ctx context.Context, workID string) (string, bool, 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) SummariesWithGeneration(ctx context.Context) ([]ingest.WorkSummary, uint64, 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 BarcodeHolder ¶ added in v0.153.0
BarcodeHolder is one item that holds a barcode: its work, its instance, and whether that work is live (not suppressed/tombstoned). Uniqueness is judged among live holders only -- a withdrawn copy's barcode may be reused.
type DuplicateBarcode ¶ added in v0.152.0
type DuplicateBarcode struct {
Barcode string `json:"barcode"`
Count int `json:"count"` // number of live items holding it
WorkIDs []string `json:"workIds"` // the works that hold it, sorted and unique
}
DuplicateBarcode is one barcode held by more than one live item in the corpus, with the works that hold it. A barcode names one physical copy, so a duplicate is a data-quality defect a librarian needs to find before uniqueness can be enforced on writes.
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) AllocateBarcodes ¶ added in v0.106.0
Barcodes returns every barcode in the corpus. The map is a copy the caller may mutate (the bulk-add generator reserves candidates in it). AllocateBarcodes runs fn holding the process-wide barcode allocation lock.
Choosing a barcode is a read-modify-write over the whole corpus: read the taken set, pick the unused ones, write them into a grain. Barcodes returns a copy and reserves nothing, so two allocators that overlap anywhere in that span choose the same numbers. The grain write cannot arbitrate it either -- two bulk adds against two different works compare-and-swap on two different objects, and both succeed.
A barcode names one physical copy, so a duplicate is not a stale read to be retried; it is the wrong label on a book. Serializing the whole span is the only thing that makes the check the caller performs mean anything.
This is correct for one process, which is the only deployment libcat supports today -- signingKey() warns at boot when a second replica would break token verification. It is the seam to replace when that changes: the allocation needs a conditional write in the shared store, per barcode, rather than a mutex in one process's memory.
Safe on a nil index: fn runs unserialized, as it did before this existed.
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) BarcodeHeldByOther ¶ added in v0.153.0
func (ix *Index) BarcodeHeldByOther(ctx context.Context, barcode, workID, instanceID string) (bool, BarcodeHolder, error)
BarcodeHeldByOther reports whether barcode is already held by a live item on a different instance than (workID, instanceID) -- the write-time uniqueness check . It excludes the given instance so re-saving that instance's own items is not a self-collision. Answers from the in-memory holders map under the index lock, with no whole-corpus copy ( sizing).
func (*Index) ClusterOwners ¶
ClusterOwners returns the works carrying the author/title/language clustering key, ordered by grain path.
func (*Index) DuplicateBarcodes ¶ added in v0.152.0
func (ix *Index) DuplicateBarcodes(ctx context.Context) ([]DuplicateBarcode, error)
DuplicateBarcodes reports every barcode held by more than one live item across the corpus, sorted by barcode. Count is the number of live item occurrences (so a barcode on two items of one work counts twice and is reported); WorkIDs is the sorted, de-duplicated set of works holding it.
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) MergedInto ¶ added in v0.148.2
MergedInto reports whether workID already records an outgoing merge -- i.e. it is a retired loser -- and, if so, the survivor it was merged into. The merge marker lives in the survivor's grain (bibframe.AddMergeMarker), not the loser's, so a merge endpoint cannot see it by reading the loser's own grain; it consults the indexed markers instead. Lets a second, contradictory merge of the same loser be refused before it writes a second marker.
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). 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.
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) -- 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) SummariesWithGeneration ¶ added in v0.119.0
SummariesWithGeneration returns every work's summary and the generation of the derived view they came from. The pair is read under one lock: a caller that read the summaries and the generation separately could cache a stale index under a fresh generation and never rebuild it. 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 . 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, 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