Documentation
¶
Overview ¶
Package publish carries approved queue decisions into the BIBFRAME grain store: editorial quads written under ETag optimistic concurrency, the advisory ingest lease, and the downstream rebuild trigger. The write discipline (read -> patch -> conditional put -> retry-from-fresh) is safe against concurrent editors and re-ingest because editorial statements are independent IRI-based quads and canonicalization dedups.
Index ¶
- Variables
- func MutateGrain(ctx context.Context, st blob.Store, path string, ...) (etag string, err error)
- type IndexUpdater
- type Lease
- func (l *Lease) Acquire(ctx context.Context, holder string) (bool, error)
- func (l *Lease) Heartbeat(ctx context.Context, holder string) error
- func (l *Lease) Held(ctx context.Context) (string, bool, error)
- func (l *Lease) Release(ctx context.Context, holder string) error
- func (l *Lease) SetClock(now func() time.Time)
- type Publisher
- type Result
Constants ¶
This section is empty.
Variables ¶
var ( ErrGrainStore = errors.New("grain store unavailable") ErrGrainConflict = errors.New("the record kept changing while it was being edited") )
ErrGrainStore and ErrGrainConflict classify the ways MutateGrain can fail that are not the caller's own doing.
MutateGrain has three error sources -- the store's read, the caller's mutate closure, and the store's write -- and used to return all three the same way. A caller could not tell "no such work", which is the user's answer and safe to show them, from an *os.PathError naming the blob root, which is neither (tasks/272). Store errors now wrap ErrGrainStore, CAS exhaustion is ErrGrainConflict, and the closure's error passes through untouched, so errors.Is sorts the three without matching on strings.
The store's own sentinel survives the wrap (blob.ErrReadOnly, and so on): callers that must tell a read-only deployment from a broken one still can, which is the discipline tasks/260 settled for the single-record route.
var ErrIngestActive = errors.New("publish: ingest lease held; publish deferred")
ErrIngestActive reports that the ingest lease is held: approvals stay queued (durable) and publishing retries after the lease expires.
Functions ¶
func MutateGrain ¶
func MutateGrain(ctx context.Context, st blob.Store, path string, mutate func(old []byte) ([]byte, error)) (etag string, err error)
MutateGrain applies mutate to the grain at path under ETag optimistic concurrency: read, transform, conditional put, retry from fresh on conflict. The exported CAS primitive record editing (tasks/037) and store-backed ingest share.
Errors sort three ways, and callers that answer a request must tell them apart before showing anything to a person (see ErrGrainStore):
- the store's read or write failed -- wraps ErrGrainStore, and the store's own sentinel too, so blob.ErrReadOnly stays reachable.
- the retries ran out -- ErrGrainConflict. It names no path: the caller knows the path and can log it; a client has no use for it.
- anything else came from mutate, unwrapped. It is the caller's own error, phrased for whoever asked.
Types ¶
type IndexUpdater ¶ added in v0.53.0
type IndexUpdater interface {
Apply(grainPath, etag string, grain []byte)
AppendFeed(ctx context.Context, paths ...string) error
}
IndexUpdater is the workindex.Index surface publish writes keep exact: Apply folds one written grain in, AppendFeed publishes the changed paths so other containers read-their-writes (mirrors batch.IndexUpdater).
type Lease ¶
type Lease struct {
// contains filtered or unexported fields
}
Lease is the advisory ingest lease: feed re-ingest holds it (identity resolution is single-flight), and the publisher defers -- never drops -- approved changes while it is held. Expiry is carried in the record body, not store TTL, because DynamoDB expires lazily.
func (*Lease) Acquire ¶
Acquire attempts to take the lease for holder. It returns false when another holder has it un-expired. Re-acquiring one's own live lease extends it.
type Publisher ¶
type Publisher struct {
Blob blob.Store
Queue *suggest.Service
Vocab *vocab.Index
// Trigger receives one grains-changed event per successful run
// (nil = no notification).
Trigger trigger.Notifier
// Lease, when set, defers publishing while ingest holds it.
Lease *Lease
// Prefix is the grain tree root in the blob store ("" = repo-layout
// paths straight from bibframe.GrainPath).
Prefix string
// Summaries, when set, is the shared maintained summary source
// (workindex, tasks/109) tag promotion scans instead of a per-run
// corpus walk; nil falls back to ScanSummaries.
Summaries ingest.SummarySource
// Index, when set, is kept exact for this publisher's own grain writes
// -- the read-your-writes contract the single-record and batch paths
// hold (tasks/195, tasks/203). Without it, tag promotions and approved
// publishes wait out the workindex refresh TTL: invisible to work
// search for up to 30s, and batch selections resolve against the
// stale index meanwhile.
Index IndexUpdater
Logger *slog.Logger
}
Publisher drains approved-unpublished queue items into grains.
func (*Publisher) PromoteTag ¶
func (p *Publisher) PromoteTag(ctx context.Context, promo suggest.Promotion, actor string) (int, error)
PromoteTag executes an approved tag promotion (tasks/044): every Work carrying the tag gains the controlled subject (with its authority labels and hierarchy), its editorial lcat:tag is retracted, and the alias grain records lcat:tagAlias so the projector suppresses the residual feed tag where the term is present and future entries auto-suggest the term. Returns the number of Works rewritten.
func (*Publisher) PublishApproved ¶
PublishApproved drains the approved-unpublished worklist: per Work, all its approved terms land in one grain mutation; each mutation stamps the queue items with the resulting grain ETag and writes an audit entry.
Source Files