internal/infra/pg/segmentstore.go
internal/infra/pg · 253 lines · 14 declarations · source
Declarations
const ProjectionSegment
const ProjectionSegment = "segment"
ProjectionSegment is the projection kind a segment registers as.
type SegmentStore
type SegmentStore struct {
pool *pgxpool.Pool
schema Schema
}
SegmentStore reads a subject's formed history and writes the segments that stand in for it.
What a subject's history is
The subject's own turns, formed and not parked, in log order. A turn that has not formed is not yet memory and may still be parked; a segment written over it would stand in for something the deployment has not admitted. A parked turn is excluded for the same reason the freshness watermark skips it: the driver gave up on it, and it is visible as parked rather than silently summarised.
func NewSegmentStore
func NewSegmentStore(pool *pgxpool.Pool, schema Schema) *SegmentStore
method SegmentStore.Subjects
func (s *SegmentStore) Subjects(ctx context.Context, scope string, limit int) ([]string, error)
Subjects lists the data subjects in a scope with more formed turns than the verbatim tail, which is every subject the pass may have something to write for.
method SegmentStore.Formed
func (s *SegmentStore) Formed(ctx context.Context, scope, subject string) ([]compaction.Turn, error)
Formed reads one subject's formed turns in log order, each with its size in characters.
method SegmentStore.Segments
func (s *SegmentStore) Segments(ctx context.Context, scope, subject string) ([]compaction.Segment, error)
Segments reads one subject's segments.
method SegmentStore.Material
func (s *SegmentStore) Material(ctx context.Context, scope, subject string, plan compaction.Plan) (compaction.Material, error)
Material loads what a planned segment is written from: the turns in its range at level 1, the summaries of its units above.
method SegmentStore.turnTexts
func (s *SegmentStore) turnTexts(ctx context.Context, scope, subject string, from, to int64) ([]compaction.TurnText, error)
var ErrSegmentConflict
var ErrSegmentConflict = errors.New("a segment already covers this range")
ErrSegmentConflict says a segment already exists where this one would go: another replica wrote it, or the plan is stale.
method SegmentStore.Write
func (s *SegmentStore) Write(ctx context.Context, scope, subject string, plan compaction.Plan, summary compaction.Summary) (compaction.Segment, error)
Write stores a segment and registers it to every observation it covers, in one transaction, so that erasure of any of them reaches it and a segment never exists without its registrations.
type Assembled
type Assembled struct {
Segments []AssembledSegment
Turns []compaction.TurnText
Characters int
Truncated bool
}
Assembled is a context as the API returns it: the segments and the turns, oldest first, with their text, and the cut.
type AssembledSegment
type AssembledSegment struct {
compaction.Segment
Summary string
}
method SegmentStore.Context
func (s *SegmentStore) Context(ctx context.Context, scope, subject string, budget int) (Assembled, error)
Context assembles one subject's history under a budget, with no model call: the planner's choice over what the store holds, then the text of what it chose.
func isUniqueViolation
func isUniqueViolation(err error) bool
isUniqueViolation says whether an error is the substrate refusing a duplicate, which for a segment is the one-per-range rule holding against a concurrent writer.