service

package
v1.0.0-alpha.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 1, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package service provides the orchestration layer for pulse operations.

Index

Constants

View Source 
const (
	// RegroupIncomingFlattened: the archive is ungrouped (0x01) and the
	// arriving shard was grouped; it was stored as its logical (0x01)
	// twin. Only the arriving shard was rewritten.
	RegroupIncomingFlattened = "incoming_flattened"
	// RegroupIncomingRegrouped: the archive is grouped and the arriving
	// shard declared a different group layout (none, or other groups);
	// it was re-encoded to the archive's layout. Only the arriving shard
	// was rewritten.
	RegroupIncomingRegrouped = "incoming_regrouped"
	// RegroupConstantPromoted: the arriving shard disagreed with a
	// CONSTANT group's one value, so that group became an indexed group
	// across the WHOLE archive — every shard re-encoded (each row gains a
	// 4-byte index).
	RegroupConstantPromoted = "constant_promoted"
)

Reasons a GroupReconciliation records.

Variables

This section is empty.

Functions

func SplitAnchorPath added in v0.8.0 

func SplitAnchorPath(path string) (string, string, bool)

SplitAnchorPath is the exported form of splitAnchorPath, used by the pulse facade (Predict, Inspect) so anchor paths resolve consistently across every entry point. See splitAnchorPath for semantics.

Types

type AddShardResult added in v0.38.0 

type AddShardResult struct {
	// Archive is the archive path that was rewritten.
	Archive string `json:"archive"`
	// Added is the source path of the shard that was appended.
	Added string `json:"added"`
	// ShardCount is the archive's shard count after the add.
	ShardCount int `json:"shard_count"`
	// Widened carries one entry per set field promoted to a wider rung;
	// empty on the ordinary append.
	Widened []SetWidening `json:"widened,omitempty"`
	// Regrouped carries one entry per parent-group layout change the add
	// made (see GroupReconciliation); empty on the ordinary append,
	// including an append whose group dictionary merely union-merged.
	Regrouped []GroupReconciliation `json:"regrouped,omitempty"`
	// Warnings carry every non-fatal diagnostic the add produced —
	// PULSE_SHARD_DESCRIPTION_DIVERGENCE from cohesion, one
	// PULSE_SHARD_SET_WIDENED per entry in Widened and one
	// PULSE_SHARD_GROUPS_REWRITTEN per entry in Regrouped. Shaped like the
	// descriptor envelope's warning entries so a caller lifts them
	// straight onto `warnings` without reshaping.
	Warnings []encoding.CohesionWarning `json:"warnings"`
}

AddShardResult is what AddShard reports.

AddShard returns a result rather than a bare error specifically so the widen warning cannot be dropped. Auto-widening is a whole-archive byte rewrite: it is far cheaper than the full re-import it replaces, but it is not free, and a caller who never learns it happened cannot tell an O(1) append from an O(archive) rewrite. The warning is mandatory, so the channel carrying it is mandatory too.

type BuildIndexResult added in v0.27.0 

type BuildIndexResult struct {
	IndexPath string
	Index     *encoding.Index

	// ManifestPath is the keyless sidecar index manifest this build
	// upserted its entry into (encoding.IndexManifestPath). Additive —
	// it names the ONE path a caller can read to discover this index's
	// key tuple without knowing the tuple already, which the sidecar's
	// own hash-derived filename makes impossible.
	ManifestPath string
}

BuildIndexResult is the outcome of a successful point-lookup sidecar index build: the derived sidecar path it was written to (via encoding.SidecarIndexPath) plus the in-memory encoding.Index that was serialized there, in case a caller wants to inspect it (or run assertions against it in a test) without a round-trip read.

type Cohort 

type Cohort struct {
	// contains filtered or unexported fields
}

Cohort represents an opened .pulse file with its parsed schema. For shard archives Shards is populated in central-directory order (which equals shard insertion order). For single-file cohorts Shards is empty.

func (*Cohort) RecordCount 

func (c *Cohort) RecordCount() (int64, error)

RecordCount returns the number of records in the cohort file. It reads the file and counts records based on the schema's per-record byte size.

func (*Cohort) Records 

func (c *Cohort) Records(_ context.Context) (processing.RecordIterator, error)

Records returns a streaming iterator over records in the cohort. Records are decoded lazily from disk — the full file is not materialized.

func (*Cohort) Schema 

func (c *Cohort) Schema() *encoding.Schema

Schema returns the cohort's schema.

func (*Cohort) Shards added in v0.8.0 

func (c *Cohort) Shards() []ShardEntry

Shards returns the shard manifest for an archive-backed cohort. Returns an empty slice for single-file cohorts. The returned slice is a defensive copy; callers may mutate freely.

type ComposeOptions added in v0.2.0 

type ComposeOptions struct {
	// MaxWorkers caps concurrent in-flight Process calls. Zero means
	// runtime.GOMAXPROCS(0). Negative values are clamped to 1.
	MaxWorkers int

	// PerRequestTimeout, if positive, derives a context.WithTimeout for
	// each request. Zero means no per-request timeout (the parent ctx's
	// deadline still applies).
	PerRequestTimeout time.Duration

	// FailFast cancels in-flight siblings on the first request error.
	// Default is true: surface errors quickly. Set false to collect every
	// request's outcome (errors aggregated into a single CodedError with
	// per-index detail).
	FailFast bool
}

ComposeOptions controls parallel execution of a ComposedRequest.

Order of responses is always preserved — slot-by-index — regardless of MaxWorkers or completion order.

type CreateShardArchiveResult added in v0.38.0 

type CreateShardArchiveResult struct {
	// Archive is the archive path that was written.
	Archive string `json:"archive"`
	// ShardCount is the number of shards the archive holds.
	ShardCount int `json:"shard_count"`
	// Widened carries one entry per set field promoted to a wider rung
	// while the archive was being assembled; empty on the ordinary
	// create.
	Widened []SetWidening `json:"widened,omitempty"`
	// Regrouped carries one entry per parent-group layout change the
	// create made while seeding (see GroupReconciliation).
	Regrouped []GroupReconciliation `json:"regrouped,omitempty"`
	// Warnings carry every non-fatal diagnostic the create produced —
	// PULSE_SHARD_DESCRIPTION_DIVERGENCE from cohesion, one
	// PULSE_SHARD_SET_WIDENED per entry in Widened and one
	// PULSE_SHARD_GROUPS_REWRITTEN per entry in Regrouped.
	Warnings []encoding.CohesionWarning `json:"warnings"`
}

CreateShardArchiveResult is what CreateShardArchive reports.

Like AddShardResult it is a RESULT rather than a bare error for one reason: the seed path auto-widens too, and the PULSE_SHARD_SET_WIDENED warning is mandatory wherever a widen happens. A create that silently re-lays-out the shards it was handed is the same failure as a silent add — the caller cannot tell a straight zip of their inputs from a rewrite of them, and the `.pulse` bytes now in the archive are not the `.pulse` bytes on disk beside it.

One rule, no asymmetry: create and add widen on the same conditions, through the same planner and executor, and report through the same SetWidening / CohesionWarning shapes.

type DecodeCallback added in v0.17.0 

type DecodeCallback func(rec *processing.Record) error

DecodeCallback is invoked once per decoded record by a parallel decode worker. Implementations MUST be safe to call from a single goroutine — each worker is given its own DecodeCallback via the DecodeCallbackFactory, so the same callback instance is never shared across workers.

Returning a non-nil error aborts the worker's decode loop; the errgroup propagates the error and cancels every other worker.

type DecodeCallbackFactory added in v0.17.0 

type DecodeCallbackFactory func(workerIdx, recordCount int) DecodeCallback

DecodeCallbackFactory constructs a per-worker DecodeCallback. Called once per worker before that worker begins decoding. The factory MUST be safe to call from the goroutine that owns workerIdx; the returned callback is consumed only by that same goroutine.

workerIdx is the worker's 0-based index (0..workers-1) and is stable for the lifetime of the parallel run — partial-state implementations can key per-worker accumulators on it. recordCount is the number of records the worker will receive (not the cohort total), useful when the callback wants to pre-size a slice or accumulator.

type GroupReconciliation added in v0.39.0 

type GroupReconciliation struct {
	// Reason is one of "incoming_flattened", "incoming_regrouped",
	// "constant_promoted".
	Reason string `json:"reason"`
	// ArchiveRewritten is true when every shard already in the archive
	// was re-encoded (constant_promoted), false when only the arriving
	// shard was.
	ArchiveRewritten bool `json:"archive_rewritten"`
	// Groups labels the groups involved: the promoted groups for
	// constant_promoted, the archive's groups otherwise.
	Groups []string `json:"groups,omitempty"`
	// ArchiveFormatVersion / IncomingFormatVersion are the .pulse format
	// versions the archive and the arriving shard declared (1 = no
	// groups, 2 = grouped).
	ArchiveFormatVersion  int `json:"archive_format_version"`
	IncomingFormatVersion int `json:"incoming_format_version"`
	// ShardsRewritten / RecordsRewritten size the rewrite.
	ShardsRewritten  int   `json:"shards_rewritten"`
	RecordsRewritten int64 `json:"records_rewritten"`
}

GroupReconciliation records one parent-group layout change a CreateShardArchive or AddShard made to fit a shard into the archive, with the size of the rewrite it cost. Typed for the same reason SetWidening is: an embedder adding shards on a schedule must be able to act on it without parsing prose. The mandatory PULSE_SHARD_GROUPS_REWRITTEN warning carries the same figures.

A plain dictionary union — an arriving shard with the archive's group layout whose tuples are renumbered into the canonical dictionary — is NOT reported here, exactly as a categorical dictionary union is not: it rewrites only the arriving shard and is the ordinary cost of an add.

type IndexFreshnessReason added in v0.27.0 

type IndexFreshnessReason string

IndexFreshnessReason names why Service.VerifyIndex reached its Fresh verdict. All three values are reachable from VerifyIndex's size fast-path / full-hash decision tree — see VerifyIndex's doc comment for the full contract. 

const (
	// IndexFreshnessReasonStatMismatch indicates the current cohort
	// file's SIZE no longer matches the sidecar's recorded
	// Index.SourceSize. The cohort has definitely changed since the
	// index was built — the fingerprint is taken over the whole file, so
	// a file of a different length cannot hash to the recorded digest —
	// and this verdict is therefore reached WITHOUT computing the
	// content hash (the size fast-path).
	//
	// A modification time that moved while the size held no longer
	// reaches this reason: mtime resolution is not preserved across
	// filesystems, so that case escalates to the fingerprint and lands
	// on IndexFreshnessReasonFingerprintMatch / _Mismatch with
	// ModTimeDrift set. See VerifyIndex's doc comment.
	IndexFreshnessReasonStatMismatch IndexFreshnessReason = "stat_mismatch"

	// IndexFreshnessReasonFingerprintMatch indicates the size matched
	// the sidecar's recorded snapshot and the full content hash
	// (recomputed as confirmation, never skipped) is byte-equal to the
	// sidecar's embedded Fingerprint. Reached whether or not the
	// modification time also matched — VerifyIndexResult.ModTimeDrift
	// reports which.
	IndexFreshnessReasonFingerprintMatch IndexFreshnessReason = "fingerprint_match"

	// IndexFreshnessReasonFingerprintMismatch indicates the size matched
	// the sidecar's recorded snapshot, but the full content hash
	// recomputed for confirmation differs from the sidecar's embedded
	// Fingerprint — either a same-size rewrite the size check cannot
	// rule out, or a same-size rewrite that also moved the mtime
	// (ModTimeDrift true).
	IndexFreshnessReasonFingerprintMismatch IndexFreshnessReason = "fingerprint_mismatch"
)

type IndexInfo added in v0.27.0 

type IndexInfo struct {
	IndexPath      string
	Keys           []string
	DistinctKeys   int
	IndexedRecords int
}

IndexInfo describes one sidecar point-lookup index discovered alongside a cohort by Service.ListIndexes: the sidecar's on-disk path (encoding.SidecarIndexPath's derived shape), its ordered key column names, and the same distinct-key / indexed-record summary Service.BuildIndex's result carries. Never re-derived by re-scanning the cohort: the counts come from the keyless index manifest when one exists (which is what lets discovery cost no sidecar read at all) and otherwise from the sidecar's own bucket table (encoding.ReadIndexFile).

type RealPather added in v0.17.0 

type RealPather interface {
	RealPath(name string) (string, error)
}

RealPather is the capability interface a filesystem implements to declare that the virtual path it serves maps to a regular on-disk file the kernel can mmap. The returned path MUST satisfy:

  • It is an absolute or process-CWD-relative path resolvable by os.Open at the moment of the call.
  • The file is a regular file (not a device, FIFO, or socket).
  • The returned bytes, when read, are byte-equal to what the filesystem's Open(name).Read would yield.

External afero.Fs implementations (e.g. a CopyOnRead overlay that caches GCS objects to a local tempdir) should satisfy this interface to opt their files into the iterator's mmap fast path. The signature intentionally matches afero.BasePathFs.RealPath so that wrapper is detected without further glue.

Returning a non-nil error declines the fast path for that call and the iterator falls back to afero.ReadFile.

type Row added in v0.2.0 

type Row = map[string]any

Row is a single result row in a processing stream. Aliased so callers can write `service.Row` without leaking the underlying map type.

type RowIter added in v0.2.0 

type RowIter interface {
	Next(ctx context.Context) (Row, bool, error)
	Close() error
	// Metadata returns the run metadata (total/filtered row counts,
	// cohort filename). Available after the iterator is exhausted; may
	// return nil before then. Streaming consumers that need metadata
	// before draining should call Process instead.
	Metadata() *types.ResponseMetadata
	// Components returns the per-operator constituent-parts metadata
	// emitted by the run — same shape as Response.Components on a
	// buffered Process call. Available after the iterator is exhausted;
	// may return nil before then or when the run produced no components
	// payload. Streaming consumers that surface per-chunk component
	// deltas read this on terminal flush to expose the buffered-only
	// payload for non-mergeable operators (AGG_MEDIAN, AGG_PERCENTILE).
	Components() *types.ResponseComponents
}

RowIter is a pull-based iterator over a processing result. Next reports (row, true, nil) for each available row, then (nil, false, nil) on exhaustion. Close releases any underlying file handles.

Implementations are NOT goroutine-safe. A single consumer per iterator.

type Service 

type Service struct {
	// contains filtered or unexported fields
}

Service is the orchestration layer connecting filesystem, encoding, and processing.

func New 

func New(fsConfig *fs.Config) *Service

New creates a new Service with the given filesystem configuration. Smart-defaults resolution runs by default; call DisableDefaults to opt out per instance (or pass pulse.Options{DisableDefaults: true} via the facade).

func (*Service) AddShard added in v0.8.0 

func (s *Service) AddShard(ctx context.Context, archivePath, shardPath string) (*AddShardResult, error)

AddShard appends shardPath to the archive at archivePath. The incoming shard is validated against the archive's canonical schema via ValidateStructuralCohesion + MergeDictUnion. If the union-merge extends the canonical dictionary, the rewritten `_schema.pulse` payload reflects the extension.

When a merged dictionary outgrows a set_* field's bitmask, the field is WIDENED to the narrowest rung that holds it — across the canonical schema and every shard payload in the archive — rather than the add being refused. Refusing forces a full re-import (re-read the source, re-infer, rebuild), which is strictly more expensive than the mechanical re-stride it would be avoiding. Widening emits a MANDATORY PULSE_SHARD_SET_WIDENED warning on the result: an expensive whole-archive rewrite that happens silently is indistinguishable from a cheap append. A union above the widest rung has nowhere to go and stays fatal (PULSE_SHARD_DICT_WIDTH_OVERFLOW).

v1 strategy: read the full archive into memory, append the new shard entry, rewrite the central directory + EOCD, and write the resulting bytes back via the atomic temp+fsync+rename pattern (§7.1). That atomicity is what makes the widen safe without locking — concurrency is caller-owned by contract, and a mid-rewrite failure leaves the original archive byte-identical and openable rather than half-widened (which would open, and decode every post-widen field at the wrong offset).

func (*Service) AutoLabels added in v0.11.2 

func (s *Service) AutoLabels() []*types.LabelBinding

AutoLabels returns the configured default bindings. Exposed for tests.

func (*Service) BuildIndex added in v0.27.0 

func (s *Service) BuildIndex(ctx context.Context, path string, keyFields []string) (*BuildIndexResult, error)

BuildIndex scans the single-file cohort at path exactly once, resolves each record's ordered-tuple composite key across every column named in keyFields to its exact on-wire bytes (processing.CompositeKeyFieldOnWireBytes — the ordered concatenation of each key column's on-wire bytes; a single-element keyFields is the degenerate 1-tuple case and byte-identical to the E1 single-key path), groups row-ids by key into hash buckets (encoding.BucketIndex / encoding.HashKey — the same hash the lookup probe uses), computes the source .pulse file's content-hash fingerprint (encoding.ComputeFingerprint), and writes the sidecar via encoding.WriteIndexFile at the path encoding.SidecarIndexPath derives. keyFields may name any number of columns >= 1; key column ORDER is significant — building on ["a","b"] and ["b","a"] produces two distinct sidecar files at two distinct derived paths, with distinct composite key byte layouts.

Row-ids are the record's 0-based sequential position in the cohort (row 0 is the first record after the header + schema), matching encoding.RecordLocator's Offset(i) addressing — lookup feeds a bucket's row-id straight into RecordLocator without any translation.

Rows where ANY key column's value is null are skipped entirely — not indexed, not a build error (processing.CompositeKeyFieldOnWireBytes returns ok=false when any component is null; a composite key with a null component is unindexable, so the whole row is skipped rather than partially indexed).

Shard archive cohorts (detected via the cheap leading-magic-bytes dispatch service.Service.Open already performs — encoding.ZipMagic vs encoding.MagicBytes, no added scan) are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED. Row-id addressing (RecordLocator.Offset) only has meaning for a single contiguous record region, which a multi-shard archive does not present. Sharded point-lookup support is out of scope for v1.

Key field types are gated by processing.IsIndexKeyableFieldType — see that predicate's doc comment for the full, settled keyable-type policy (E2-S3). Rejected types (currently only set_*) fail fast here with PROCESSING_CONFIG and processing.IndexKeyRejectionMessage's type-specific explanation, before any record is scanned.

func (*Service) CompactShardArchive added in v0.8.0 

func (s *Service) CompactShardArchive(ctx context.Context, archivePath string) error

CompactShardArchive rewrites the archive at archivePath to eliminate orphaned bytes from prior in-place mutations and refresh the canonical schema-doc metadata (aggregate_record_count + shard_count). The rewrite reads every shard payload (excluding the reserved `_schema.pulse` entry) in central-directory order, re-emits the canonical schema with the recomputed metadata, and writes the result via the standard atomic temp+rename pattern (§7.1).

v1 note: AddShard / RemoveShard already use temp+rename for the whole-archive rewrite, so v1 archives never accumulate orphan bytes from Pulse-issued mutations. Compact still has a job in v1: it refreshes the canonical _schema.pulse metadata that may have drifted (for example, if the archive was edited by a third-party zip tool outside Pulse) and is the explicit reclaim path for archives containing orphan bytes from any source. Future PRs may introduce true in-place AddShard atop lower-level zip primitives; Compact's contract remains identical — produce a clean archive byte-for-byte.

Errors:

  • PULSE_ARCHIVE_MAGIC_INVALID / PULSE_ARCHIVE_CORRUPT — surfaced when the source archive cannot be opened.
  • PULSE_SHARD_HEADER_INVALID — surfaced when a shard payload's header cannot be parsed during record-count recomputation.
  • SERVICE_RESOURCE — I/O failure on read, temp-write, or rename.

func (*Service) Compose 

func (s *Service) Compose(ctx context.Context, composed *types.ComposedRequest) (*types.ComposedResponse, error)

Compose executes multiple requests, returning a response for each.

Before dispatching any slot, the validate phase synthesizes a default Label of `request_<index+1>` (1-based) on every slot whose caller- supplied Label is empty, against an in-memory clone of each *Request so the caller's pointer is not mutated. Two slots resolving to the same final Label are rejected with PULSE_COMPOSE_LABEL_COLLISION. Compose-only overlay kinds resolve sibling references by final Label so the names must be unique across the batch.

func (*Service) ComposeParallel added in v0.2.0 

func (s *Service) ComposeParallel(
	ctx context.Context,
	composed *types.ComposedRequest,
	opts ComposeOptions,
) (*types.ComposedResponse, error)

ComposeParallel runs every request in composed concurrently across a bounded worker pool. Responses are returned in the same order as composed.Requests; per-request errors are surfaced according to opts.

Registry factories return fresh stateful instances per request, so concurrent Process calls do not share aggregator/attribute state. Geo and decimal aggregators dispatch through buffered code paths that are also safe for concurrent invocation (no shared mutable state).

func (*Service) CountRecords added in v0.10.0 

func (s *Service) CountRecords(ctx context.Context, path string) (uint64, error)

CountRecords returns the total record count for the cohort at path without decoding any payload bytes. Three resolution paths:

  • Single-file `.pulse`: stat the file, stream the header + schema, and derive count = (file_size − header − schema) / record_stride. Bytes read is bounded by header + schema size — O(1) in the record count.
  • Shard archive (zip magic): read the archive's central directory
  • the reserved `_schema.pulse` entry. The SHRD trailer's AggregateRecordCount is the cached sum of every shard's record count; cohesive archives keep it in lockstep with the per-shard headers. Cost is O(N shards) for the directory walk, not O(total records).
  • Anchor path `archive#shard`: opens the archive's central directory and uses the named shard's header to derive count. Cost is O(shard payload size) today; the v1 contract is "no full-archive decode," not "header-only" for anchors.

Embedders that want strict O(1) for anchors can call pulse.Inspect(archive) and read the per-shard counts from the returned Shards slice.

func (*Service) CreateShardArchive added in v0.8.0 

func (s *Service) CreateShardArchive(ctx context.Context, archivePath string, shardPaths []string) (*CreateShardArchiveResult, error)

CreateShardArchive builds a fresh Pulse shard archive at archivePath from the constituent shardPaths. The first shard seeds the canonical schema; every other shard is validated via encoding.ValidateStructuralCohesion + encoding.ValidateDictPrefixRule. Dictionary growth across shards accumulates into the canonical schema. The archive is written to a temp path on the same filesystem and Rename'd over the destination — partial writes never appear at archivePath (atomic temp+rename, §7.1).

Set-width AUTO-WIDEN applies here exactly as it does on AddShard: a set column whose merged dictionary outgrows its declared bitmask, or whose rung differs between two of the shards being seeded, is promoted to the narrowest rung that holds every member, across the canonical schema and every shard accumulated so far. Refusing at create while auto-widening at add was an asymmetry with no defence — the same two files produce an archive or an error depending only on whether they arrived together or one after the other.

Widening emits a MANDATORY PULSE_SHARD_SET_WIDENED warning on the result, which is why this returns a *CreateShardArchiveResult rather than a bare error. A silent whole-archive rewrite is exactly the failure class the byte-layout contract exists to prevent: the shards stored in the archive are no longer byte-identical to the `.pulse` files the caller named, and nothing else would say so.

A union above the widest rung has nowhere to go and stays fatal (PULSE_SHARD_DICT_WIDTH_OVERFLOW).

Errors:

  • PULSE_SHARD_RESERVED_NAME — any shardPath has basename `_schema.pulse`.
  • PULSE_SHARD_NAME_COLLISION — two shardPaths share a basename.
  • PULSE_SHARD_SCHEMA_MISMATCH / PULSE_SHARD_DICT_DIVERGENCE — propagated from cohesion validators.
  • PULSE_SHARD_DICT_WIDTH_OVERFLOW — a set union past the widest rung.
  • SERVICE_VALIDATION — empty shardPaths.
  • SERVICE_RESOURCE — I/O failure on any input/output.

func (*Service) DecodeWorkers added in v0.17.0 

func (s *Service) DecodeWorkers() int

DecodeWorkers returns the configured cap. Exposed for tests and the buffered-decode orchestrator.

func (*Service) DedupPreflight added in v0.39.0 

func (s *Service) DedupPreflight(_ context.Context, path string) error

DedupPreflight performs the layout dispatch a retro-dedup needs before the byte work starts (io.DedupJob, which speaks only the single-file layout). It mirrors WidenSetField's dispatch exactly:

  • SERVICE_VALIDATION — path is an anchored shard (`archive.pulse#shard.pulse`): a shard inside an archive cannot be regrouped alone, because every shard shares the canonical schema.
  • SERVICE_VALIDATION — path is a shard archive (zip magic). details.layout = "shard_archive". Shard archives DO carry grouped shards (shard create/add union-merge their dictionaries), but an archive-wide dedup is a different operation from a per-file one: the viability gate, constant elision and group suggestion would have to be decided over the union of every shard (a cross-shard observation pass), and every shard rewritten atomically to one layout. That is not built; the supported route is to dedup each source shard (or `shard extract` + dedup) and `shard create` the results, which reconciles them to one layout.
  • SERVICE_RESOURCE — the cohort does not exist / cannot be opened.

A nil return means path is a single-file cohort the job may read.

func (*Service) DisableComponents added in v0.22.2 

func (s *Service) DisableComponents() bool

DisableComponents reports the engine-level setting.

func (*Service) DropIndex added in v0.27.0 

func (s *Service) DropIndex(ctx context.Context, path string, keyFields []string) error

DropIndex removes the sidecar point-lookup index built for keyFields against the cohort at path (encoding.SidecarIndexPath derives the exact path). Non-interactive by design — there is no confirmation prompt (a prompt would block a scripted/automated caller); callers that want a confirm-before-delete UX build it on top of ListIndexes + DropIndex themselves. Sidecars are cheap, deterministic rebuild artifacts (see encoding.IndexFormatVersion's doc comment), so dropping one is never destructive of anything Service.BuildIndex cannot regenerate.

The cohort's keyless sidecar index manifest (encoding.IndexManifestSuffix) has the corresponding entry pruned in the same call, so a dropped index stops being discoverable as well as stopping being readable.

Returns PULSE_INDEX_MISSING (not a silent no-op) when neither a sidecar NOR a manifest entry exists at the derived path for keyFields — a caller who asked to drop a specific key-set's index gets a clear, coded signal rather than an ambiguous success, matching Service.VerifyIndex's contract for the same "no sidecar found" case. An ORPHANED manifest entry (the entry is present, the file is gone — an `rm` behind Pulse's back) is a successful drop rather than PULSE_INDEX_MISSING: that state is what PULSE_INDEX_MANIFEST_STALE reports, and this is the call its fixup points at. Shard archive cohorts are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED, matching Service.BuildIndex / Service.VerifyIndex / Service.ListIndexes's v1 single-file-only contract.

func (*Service) EchoRequest added in v0.13.0 

func (s *Service) EchoRequest() bool

EchoRequest reports the current echo setting.

func (*Service) Extensions added in v0.7.0 

func (s *Service) Extensions() *processing.ExtensionRegistry

Extensions returns the installed ExtensionRegistry, or nil when no extensions are registered. Exposed for descriptor/manifest emission and MCP schema-binding paths.

func (*Service) ExtensionsSnapshot added in v0.7.0 

func (s *Service) ExtensionsSnapshot() *descriptor.ExtensionsSnapshot

ExtensionsSnapshot returns the descriptor-side projection of the registered extensions, or nil when no extensions are installed.

func (*Service) ExtractShard added in v0.8.0 

func (s *Service) ExtractShard(ctx context.Context, archivePath, shardBasename string) (io.ReadCloser, error)

ExtractShard returns an io.ReadCloser over the named shard's standalone single-file `.pulse` bytes. The closer releases the archive resources; callers must invoke Close after draining.

func (*Service) Facet 

func (s *Service) Facet(ctx context.Context, path string, field string) ([]string, error)

Facet returns distinct values for the named field in the cohort.

For categorical fields the dictionary is the authoritative source:

  • Single-file cohorts return the schema's field dictionary directly.
  • Shard archives return the canonical dictionary carried by `_schema.pulse`, which is guaranteed to be the union of every shard's per-shard dictionary by the append-only prefix rule enforced at insert time (sharding design contract §3.2). No record streaming is required.

For numeric fields the cohort is streamed once and the set of distinct values observed is returned in encounter order. For shard archives the stream spans every shard in central-directory (insertion) order — the union semantics specified in §5.4.

func (*Service) FacetSchema added in v0.7.0 

func (s *Service) FacetSchema(ctx context.Context, req *types.FacetRequest) (*types.FacetResult, error)

FacetSchema is the rich multi-field facet implementation. The pulse.FacetSchema facade is a thin wrapper around this method.

Algorithm: one streaming pass over the cohort applies the base filterers, then for each requested field updates the appropriate accumulator (discrete dictionary count, online numeric stats, online histogram bin). Numeric percentiles force a buffered second-stage sort over the per-field non-null value slices. Additive contribution counts run a parallel discrete accumulator with the additive field's own filter clauses stripped from the base filter.

func (*Service) FilterToFile added in v0.8.1 

func (s *Service) FilterToFile(ctx context.Context, src, dst, filterExpr string) (int64, error)

FilterToFile reads the .pulse cohort at src, evaluates filterExpr against every record using FILTER_EXPRESSION semantics, and writes a new .pulse cohort at dst containing only records that pass. Custom expr functions and lookup tables registered via pulse.Options.Extensions are honored.

Three input shapes are supported, dispatched on the first four bytes:

  • Single-file cohort (magic "PULSE\x00\x00\x00"): the dst is a single-file cohort whose header + schema bytes are copied byte-for-byte from src; only the record payload differs. A grouped (0x02) source therefore yields a 0x02 destination carrying the source's group descriptors and dictionaries UNCHANGED — surviving physical rows are copied by reader position, so every row's entry index stays valid. Entries no surviving row references are carried, not pruned (a deliberate v1 choice: pruning costs a pass and renumbers indices); inspect's ratio is then surviving rows / carried entries. Re-deduping the output with the same groups prunes. A filter that keeps nothing writes the preamble alone: a valid, empty 0x02 cohort. Records are handed their row's entry indices (RecordReader.GroupIndex), so a filter over one group's members takes the per-entry precompute.

  • Shard archive (magic "PK\x03\x04"): the dst is a new shard archive preserving the per-shard layout — one input shard maps to one output shard at the same basename, in central-directory (insertion) order. Per-shard header + schema + categorical dictionary bytes are copied byte-for-byte (records inside each shard reference that shard's local dictionary, which is preserved under the prefix-rule invariant). Shards with zero surviving records are written as empty payloads (header + schema + zero record bytes). The canonical `_schema.pulse` trailer is refreshed: aggregate_record_count = sum of surviving rows across shards, shard_count = unchanged.

  • Anchor syntax `archive.pulse#shard.pulse`: resolves to a single shard inside an archive; dst is a single-file cohort built from that shard alone.

Returns the number of records written to dst (sum across shards for archive inputs).

func (*Service) FilterToFileBySetAndExpr added in v0.8.3 

func (s *Service) FilterToFileBySetAndExpr(ctx context.Context, src, dst, includeField string, set processing.MemberSet, filterExpr string) (int64, error)

FilterToFileBySetAndExpr applies an include-set predicate (membership test against an externally-supplied list of values) and optionally a filter expression to every record in src, writing the survivors to dst. At least one of (set, filterExpr) must be supplied — if both, the set is tested first and the expression evaluated only on surviving rows (cheap-check-first short-circuit).

Input-shape dispatch matches FilterToFile (single-file, shard archive, anchor). includeField names the field whose value is tested for membership in set; the set must have been built against the same canonical schema as src (the field is resolved per-call to keep the predicate schema-coherent across shards).

Returns the number of records written to dst.

func (*Service) ListIndexes added in v0.27.0 

func (s *Service) ListIndexes(ctx context.Context, path string) ([]IndexInfo, error)

ListIndexes enumerates every sidecar point-lookup index built against the cohort at path, and is the answer to "what key tuples can I look this cohort up by" — a question the sidecar naming scheme otherwise makes unanswerable, since encoding.SidecarIndexPath derives each sidecar's filename from a HASH of its ordered key tuple, so the file cannot be opened without already knowing its contents.

TWO SOURCES, MANIFEST FIRST:

  1. The keyless manifest (encoding.IndexManifestSuffix, "cohort.pulse.indexes.json") when one exists. It carries every entry's ordered key tuple AND its distinct-key / indexed-record counts, so this path reads NO sidecar bucket table at all: one read of the document plus one existence probe per entry. That is what makes discovery work on object storage, which cannot list a directory (and is why Pulse writes the manifest at all).
  2. The directory listing (afero.ReadDir, never raw os), keeping entries whose basename matches the SidecarIndexPath shape for this cohort and reading each match's key spec + bucket table (encoding.ReadIndexFile).

The two UNION when both are available: the manifest is the guaranteed floor, and a sidecar on disk the manifest does not name (built by a Pulse predating the manifest, then never rebuilt) is still listed rather than disappearing the day a manifest appears beside it. When a manifest exists, a failure to list the directory is NOT an error — the manifest already answered, and a backend with no listing is the case it was written for.

A manifest entry naming a sidecar that is not present is PULSE_INDEX_MANIFEST_STALE rather than a silently skipped entry; see listIndexesFromManifest. A malformed manifest is PULSE_INDEX_MANIFEST_INVALID rather than a silent fall back to the directory listing, which would succeed locally and answer "no indexes" on a bucket.

Results are sorted by IndexPath for a deterministic, rebuild-stable listing order (a directory listing's raw order is not guaranteed across afero backends).

No sidecar indexes is not an error — ListIndexes returns an empty (non-nil) slice. Shard archive cohorts (same leading-magic-bytes dispatch Service.Open already performs) are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED, matching Service.BuildIndex / Service.VerifyIndex's v1 single-file-only contract.

func (*Service) ListShards added in v0.8.0 

func (s *Service) ListShards(ctx context.Context, archivePath string) ([]ShardEntry, error)

ListShards opens the archive and returns its ShardEntry slice in central-directory order (reuses Service.Open which peeks per-shard headers to populate RecordCount). Single-file cohorts return an empty slice.

func (*Service) Lookup added in v0.27.0 

func (s *Service) Lookup(ctx context.Context, req *types.LookupRequest) (*types.LookupResult, error)

Lookup resolves a point lookup against the cohort named in req.Cohort, keyed on the ordered tuple req.KeyComponents() returns (either req.Keys verbatim, for a composite/multi-column key, or the req.Field/req.Value single-key convenience path — see types.LookupRequest.KeyComponents): derive the sidecar index path for the ordered key-field names (the same derivation Service.BuildIndex used to write it), read only the sidecar's cheap bucket-data-free prefix (encoding.ReadIndexMeta) to validate the request's key-component count and run the read-path staleness check, resolve every component's literal to on-wire key bytes and concatenate them in order, seek directly to and parse the ONE hash bucket the key resolves to (encoding.ReadBucketByKey) for a byte-equal entry, and — on a hit — read the matching record(s) via encoding.RecordLocator .ReadRecordAt with a single reused DecodePlan projected to req.ReturnColumns, seeking straight to each matched record's byte offset rather than decoding the cohort from the start.

This is an O(1) operation in the cohort's record count: neither the sidecar index nor the cohort payload is ever read in full. Only the bucket-data-free index prefix, the one resolved bucket, and the matched record(s) (plus the fixed header+schema prefix each file carries) are read.

Key column order is significant: KeyComponents in a different order derives a different sidecar path and a different composite key byte layout, even when the same set of columns/values is supplied.

Multiplicity: when the matched bucket entry carries more than one row-id (a duplicate key value in the source cohort), req.Multiplicity selects the behavior — see types.LookupMultiplicity's doc comment. The zero value (unset) and LookupMultiplicityAssertUnique both mean "assert unique": Lookup fails with PULSE_LOOKUP_AMBIGUOUS when more than one row-id matches. LookupMultiplicityFirst takes the lowest row-id deterministically; LookupMultiplicityAll returns every matching row. Row order (both "first"'s pick and "all"'s slice order) is always ascending row-id — the sidecar bucket's RowIDs slice is not itself guaranteed sorted, so Lookup sorts a copy before use.

Shard archive cohorts (detected via the cheap leading-magic-bytes dispatch Service.Open already performs) are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED — sharded point-lookup is out of scope for v1. See Service.BuildIndex's doc comment for why.

func (*Service) Open 

func (s *Service) Open(ctx context.Context, path string) (*Cohort, error)

Open reads a .pulse file and returns a Cohort with the parsed schema. Dispatches on the file's leading magic bytes:

  • "PULSE\x00\x00\x00" (single-file cohort) — reads header + schema directly from the file. Shards is empty.
  • "PK\x03\x04" (zip archive) — parses the zip central directory, reads the canonical schema from the reserved `_schema.pulse` entry, and populates Shards with every other entry in central-directory order. RecordCount is populated from `_schema.pulse` metadata or shard headers.

Returns PULSE_ARCHIVE_MAGIC_INVALID when the file matches neither magic, PULSE_ARCHIVE_CORRUPT when the zip EOCD or central directory is unreadable, and PULSE_SHARD_MISSING when an archive lacks the reserved `_schema.pulse` entry.

func (*Service) OpenAnchor added in v0.8.0 

func (s *Service) OpenAnchor(_ context.Context, archivePath, entry string) (*Cohort, error)

OpenAnchor opens a single shard inside a Pulse shard archive as if it were a standalone single-file `.pulse` cohort. The schema is read from the shard's own header (NOT the archive's canonical schema) so anchor-addressed shards stand alone for the purposes of facade methods. Shards is empty on the returned Cohort.

Errors:

  • PULSE_ARCHIVE_MAGIC_INVALID — archivePath is not a Pulse shard archive.
  • PULSE_SHARD_MISSING — the named entry is not present in the archive.
  • PULSE_SHARD_RESERVED_NAME — entry is the reserved `_schema.pulse`.
  • PULSE_SHARD_HEADER_INVALID / ENCODING_INVALID — the shard payload is malformed.

func (*Service) Process 

func (s *Service) Process(ctx context.Context, req *types.Request) (*types.Response, error)

Process executes a single request against the specified cohort. Records are streamed from disk — the full file is never held in memory as raw bytes alongside the decoded records.

func (*Service) ProcessChain added in v0.10.0 

func (s *Service) ProcessChain(ctx context.Context, req *types.ChainRequest) (*types.ChainResponse, error)

ProcessChain executes a source-rooted linear chain of Process requests. The first stage runs against the cohort identified by req.Cohort; each subsequent stage receives the previous stage's rows as its input, materialised through an in-memory SliceIterator against a synthesised schema.

All stages must pass processing.CanChainRequest (mergeable + scalar-emitting aggregators only). The executor returns PULSE_CHAIN_NOT_MERGEABLE with the offending stage index in details on first failure, allowing callers to fall back to per-stage Process. Stage 0's Request.Cohort is replaced with req.Cohort; any Request.Cohort on stages >= 1 is ignored.

func (*Service) ProcessStream added in v0.2.0 

func (s *Service) ProcessStream(ctx context.Context, req *types.Request) (RowIter, error)

ProcessStream executes a request and returns a pull-based row iterator over the result. The semantics match Process — same gates, same errors, same metadata — but the consumer can drain rows incrementally instead of receiving the whole []map[string]any up front.

Today the iterator buffers internally (calls Process and walks Data). The signature is forward-compatible with a future true-streaming path driven by the streaming Processor; consumers that adopt it now will pick up that change without a code change on their side.

func (*Service) ProjectBufferedFields added in v0.8.0 

func (s *Service) ProjectBufferedFields() bool

ProjectBufferedFields reports the current setting. Read by callers that need to know whether to compute NeededFields ahead of time.

func (*Service) RemoveShard added in v0.8.0 

func (s *Service) RemoveShard(ctx context.Context, archivePath, shardBasename string) error

RemoveShard rewrites the archive at archivePath omitting the named shard. The canonical schema is preserved (orphaned dictionary entries are retained — canonical only grows). Returns PULSE_SHARD_MISSING when the named shard is not present.

func (*Service) ResolveCanonicalSchema added in v0.8.3 

func (s *Service) ResolveCanonicalSchema(_ context.Context, src string) (*encoding.Schema, error)

ResolveCanonicalSchema reads the canonical schema for src without streaming the record payload. Single-file cohorts return their own schema; shard archives return the canonical `_schema.pulse` schema; anchored shards return the anchored shard's local schema. Used by callers (notably the CLI cohort-filter include-from path) that need schema introspection to build an include-set before invoking FilterToFileBySetAndExpr.

func (*Service) Sample 

func (s *Service) Sample(ctx context.Context, path string, n int) ([]map[string]any, error)

Sample returns up to n rows from the cohort as maps of field name to value. Streams from disk — stops reading as soon as n rows are collected. For shard archives the rows span the union of shards in central-directory (insertion) order; offset+limit apply globally across the union, never per-shard (sharding design contract §5.4).

func (*Service) SampleWithRequest added in v0.10.1 

func (s *Service) SampleWithRequest(ctx context.Context, req *types.SampleRequest, path string) ([]map[string]any, []processing.ResolverWarning, error)

SampleWithRequest is the labelled variant. The req struct carries the cohort path (via Cohort.Filename), the row cap (N), and any LabelBinding entries that translate categorical values to display labels in the returned rows. Path resolution mirrors the rest of the facade — Cohort.Filename is resolved against the Service's data-dir / fs. Returns the rows plus the resolver-side warnings (PULSE_LABEL_COLLISION / PULSE_LABEL_LOOKUP_MISS) so callers can fold them into the response envelope at their boundary.

func (*Service) SetAutoLabels added in v0.11.2 

func (s *Service) SetAutoLabels(bindings []*types.LabelBinding)

SetAutoLabels installs the default LabelBindings injected into read requests before validation. Pass nil to clear. See the autoLabels field and applyAutoLabels for the schema-aware filtering applied per request. Matches pulse.Options.AutoLabels.

func (*Service) SetDecodeWorkers added in v0.17.0 

func (s *Service) SetDecodeWorkers(n int)

SetDecodeWorkers configures the per-cohort parallel decode worker pool cap used by the buffered Process path. Zero falls back to runtime.NumCPU() at dispatch time when the cohort exceeds parallelDecodeRecordThreshold; 1 disables parallelism (strictly serial path). Negative values are not rejected here — pulse.New() performs that validation at the public API boundary.

func (*Service) SetDisableComponents added in v0.22.2 

func (s *Service) SetDisableComponents(disabled bool)

SetDisableComponents toggles the engine-level default for Response.Components emission. When true, every Process / Compose / ProcessChain / Facet response on this service skips the Components block unless the per-request Request.DisableComponents pointer forces the opposite decision. See pulse.Options.DisableComponents for the full contract.

func (*Service) SetDisableCrosstabFusion added in v0.17.0 

func (s *Service) SetDisableCrosstabFusion(disabled bool)

SetDisableCrosstabFusion toggles the fused-crosstab dispatch in processCrosstab. When true, every crosstab request takes the buffered RunCrosstab path even when CanFuseCrosstab would accept it. Exposed so the equivalence test in service/crosstab_fused_test.go can drive the same request through both paths and assert byte-equal output without relying on internals.

func (*Service) SetDisableDefaults added in v0.5.0 

func (s *Service) SetDisableDefaults(disabled bool)

SetDisableDefaults toggles the smart-defaults pass. Predict still computes and reports DefaultsApplied independently; this flag governs only what the runtime mutates before Process / Compose execution.

func (*Service) SetEchoRequest added in v0.13.0 

func (s *Service) SetEchoRequest(enabled bool)

SetEchoRequest toggles per-stage normalized-request capture in ProcessChain. When true, ChainResponse carries a NormalizedRequest snapshot built during execution; when false, that field is nil. Matches pulse.Options.EchoRequest.

func (*Service) SetExtensions added in v0.7.0 

func (s *Service) SetExtensions(r *processing.ExtensionRegistry)

SetExtensions installs an ExtensionRegistry containing embedder- registered operator overlays. The registry is read-only after this call; pass nil to clear. The processor consults this registry before falling through to built-in factories.

func (*Service) SetExtensionsSnapshot added in v0.7.0 

func (s *Service) SetExtensionsSnapshot(snap *descriptor.ExtensionsSnapshot)

SetExtensionsSnapshot installs the descriptor-side projection of the registered extensions for manifest + predict consumption. Pass nil to clear; pulse.New populates this alongside SetExtensions.

func (*Service) SetProjectBufferedFields added in v0.8.0 

func (s *Service) SetProjectBufferedFields(enabled bool)

SetProjectBufferedFields enables buffered-decode field projection. When enabled the streaming iterator computes the set of fields a request actually reads (NeededFields) and skips map writes for fields outside that set. Per-record memory drops proportional to the projection ratio; decode CPU is unchanged. Default is false; pulse.Options{ProjectBufferedFields: true} opts in.

Extension operators without a registered FieldInputs hook widen the projection to "every field" automatically — projection is always a strict superset of the fields actually consumed, so it can never produce a wrong answer.

func (*Service) SetShardWorkers added in v0.8.0 

func (s *Service) SetShardWorkers(n int)

SetShardWorkers configures the per-shard parallel worker pool cap. Zero falls back to runtime.NumCPU() at dispatch time; 1 disables parallelism (strictly serial path). Negative values are not rejected here — pulse.New() performs that validation at the public API boundary.

func (*Service) SetStrict added in v0.8.4 

func (s *Service) SetStrict(strict bool)

SetStrict toggles the strict request-validation flag. When true, Process promotes the categorical-aggregation warning into a PULSE_AGG_NOT_MEANINGFUL_FOR_CATEGORICAL CodedError instead of running the request.

func (*Service) ShardWorkers added in v0.8.0 

func (s *Service) ShardWorkers() int

ShardWorkers returns the configured cap. Exposed for tests and the shard-reduce orchestrator.

func (*Service) Strict added in v0.8.4 

func (s *Service) Strict() bool

Strict reports the current strict-validation setting.

func (*Service) VerifyIndex added in v0.27.0 

func (s *Service) VerifyIndex(ctx context.Context, path string, keyFields []string) (*VerifyIndexResult, error)

VerifyIndex reports whether the sidecar point-lookup index built for keyFields against the cohort at path is still fresh, using a size fast-path before paying for a full content-hash recompute:

  1. Load the sidecar's metadata only (encoding.ReadIndexMetaFile at the path encoding.SidecarIndexPath derives — never the bucket data, which freshness never needs). PULSE_INDEX_MISSING if absent.
  2. Stat the CURRENT cohort file and compare its SIZE against the sidecar's recorded Index.SourceSize (captured by Service.BuildIndex at build time). If it differs, the cohort has definitely changed — the fingerprint covers the whole file, so a different length cannot produce the recorded digest — so report Fresh=false / Reason=IndexFreshnessReasonStatMismatch / FastPath=true WITHOUT computing the content hash.
  3. If the size matches, that alone is not sufficient proof of freshness (a pathological same-size rewrite admits a false negative), so fall through to a full encoding.ComputeFingerprint recompute and compare against the sidecar's embedded Fingerprint for a conclusive verdict (IndexFreshnessReasonFingerprintMatch / IndexFreshnessReasonFingerprintMismatch, both FastPath=false). A modification time that moved while the size held takes this same path and sets ModTimeDrift: mtime resolution is NOT preserved across filesystems (the snapshot is Unix nanoseconds, an S3 LastModified is whole seconds, and every copy and cache hop truncates somewhere), so drift alone is not evidence of mutation and the fingerprint is what answers.

VerifyIndex never rebuilds the sidecar and never falls back to a full cohort scan — it only reports freshness. Rebuilding stays an explicit, caller-initiated Service.BuildIndex call (surfaced as `pulse index build`). This method underpins the `pulse index verify` CLI leaf (a later story).

Shard archive cohorts (detected via the cheap leading-magic-bytes dispatch Service.Open already performs) are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED — sharded point-lookup is out of scope for v1. See Service.BuildIndex's doc comment for why.

func (*Service) VerifyShardArchive added in v0.8.0 

func (s *Service) VerifyShardArchive(ctx context.Context, archivePath string) (*VerifyResult, error)

VerifyShardArchive walks the archive at archivePath and re-validates every shard against the canonical schema. For every real shard (excluding `_schema.pulse`) it:

  1. Calls encoding.Archive.PeekShardHeader to check magic + format_version, raising PULSE_SHARD_HEADER_INVALID on mismatch.
  2. Parses the shard's full header + schema block.
  3. Calls encoding.ValidateStructuralCohesion against the canonical schema. Structural mismatches surface as PULSE_SHARD_SCHEMA_MISMATCH errors; per-field description divergence emits PULSE_SHARD_DESCRIPTION_DIVERGENCE warnings. Structural cohesion includes the parent-group layout: a shard whose groups differ from the canonical schema's (a 0x01 shard in a 0x02 archive, or the reverse) has a different physical stride and is refused.
  4. Calls encoding.ValidateDictPrefixRule. Any prefix-rule violation — categorical, set, or a parent group's entries — surfaces as PULSE_SHARD_DICT_DIVERGENCE.
  5. Reports per-set-field width headroom and per-group index headroom on the result, so an impending `shard add` auto-widen, a constant promotion or a u32 overflow is foreseeable.
  6. Re-peeks the shard's record count and sums across the archive. A drift against the canonical AggregateRecordCount emits a warning (the live per-shard sum is authoritative per the design contract).

Returns a non-nil error only when the archive itself cannot be opened (PULSE_ARCHIVE_MAGIC_INVALID, PULSE_ARCHIVE_CORRUPT, SERVICE_RESOURCE). Per-shard cohesion failures are accumulated into VerifyResult.Errors so the caller renders the full diagnosis — partial failures should not mask later shards' issues.

func (*Service) WidenSetField added in v0.38.0 

func (s *Service) WidenSetField(_ context.Context, path, field string, target encoding.FieldType) (*encoding.WidenReport, error)

WidenSetField widens the set column named field in the single-file cohort at path to the wider set rung target, rewriting the cohort in place, and returns the report describing what moved.

The byte work is entirely encoding.WidenSetFieldFile: it builds the re-strided cohort into a temp file beside the original, fsyncs it and renames it over the original, so a failure at any point leaves the original byte-identical. What this layer adds is the LAYOUT DISPATCH a `.pulse` path needs and the engine deliberately does not perform — the extension covers two formats (single-file, PULSE magic; shard archive, zip magic PK\x03\x04) and the widen engine speaks only the first. An archive handed to it is refused by ReadHeader before a byte is written, so the archive survives, but the caller is told the file is not a cohort rather than told what is actually true: that widening an archive is a different operation, performed across the canonical `_schema.pulse` and every shard payload at once. This method names that distinction instead.

Refusals are coded errors, each reusing an existing code:

  • SERVICE_RESOURCE — the cohort does not exist / cannot be opened.
  • SERVICE_VALIDATION — the path is a shard archive, or an anchored shard inside one (`archive.pulse#shard.pulse`).
  • ENCODING_INVALID — no field of that name in the schema.
  • ENCODING_TYPE_MISMATCH — the field is not a set, the target is not wider, or the field is already at the widest rung (encoding.CheckSetWiden owns that whole policy).

Sidecars are not rebuilt; see encoding.WidenSetFieldFile for why a widened cohort invalidates them through their own fingerprints.

type SetWidening added in v0.38.0 

type SetWidening struct {
	// Field is the widened set field's name.
	Field string `json:"field"`
	// From is the rung the archive declared before the rewrite.
	//
	// From == To is the one-shard case: the arriving shard declared a
	// NARROWER rung than the archive and was promoted to meet it, so no
	// existing shard moved. Callers deciding whether an archive-wide
	// rewrite happened compare From against To — never the presence of
	// the entry, which covers both directions.
	From string `json:"from"`
	// To is the rung every shard in the archive declares after the
	// rewrite. Never narrower than From or IncomingFrom.
	To string `json:"to"`
	// IncomingFrom is the rung the ARRIVING shard declared. It differs
	// from From exactly when the two sides disagreed on the rung, which
	// is the case a re-import of a new period produces: the source
	// inferred the rung its own data needed.
	IncomingFrom string `json:"incoming_from"`
	// DictEntries is the merged dictionary size the two schemas imply.
	DictEntries int `json:"dict_entries"`
	// ShardsRewritten counts every shard payload actually re-laid-out.
	// An archive widen rewrites the archive's existing shards plus the
	// incoming one, because after it no shard carries the old stride; a
	// promotion of a narrow incoming shard rewrites exactly that one.
	ShardsRewritten int `json:"shards_rewritten"`
	// RecordsRewritten is the total number of records re-laid-out
	// across those shards.
	RecordsRewritten int64 `json:"records_rewritten"`
}

SetWidening records one set field promoted to a wider rung while a shard was being added, together with the size of the rewrite it cost.

It is TYPED rather than only narrated in the warning message because an embedder that adds shards on a schedule needs to act on the fact (rebuild the point-lookup index, re-export, alert) without parsing prose. The mandatory warning carries the same figures for a human or an LLM reading the `--json` envelope.

func (SetWidening) ArchiveWidened added in v0.38.0 

func (w SetWidening) ArchiveWidened() bool

ArchiveWidened reports whether this widening promoted the ARCHIVE — the canonical schema and every shard already stored in it — rather than only promoting the arriving shard to meet a rung the archive already declared. The two cost wildly different amounts and the warning wording differs accordingly.

type ShardEntry added in v0.8.0 

type ShardEntry struct {
	// Filename is the basename of the shard inside the archive
	// (e.g. "20190101.pulse"). Equals the zip entry name.
	Filename string

	// RecordCount is the number of records carried by the shard.
	// Populated either from the `_schema.pulse` aggregate metadata or
	// from a first-read peek of the shard's own header.
	RecordCount int64
}

ShardEntry is one shard inside a Pulse shard archive. Populated on a Cohort opened against a `.pulse` artifact whose first four bytes match the zip magic; empty for single-file cohorts.

type VerifyIndexResult added in v0.27.0 

type VerifyIndexResult struct {
	IndexPath string
	Fresh     bool
	Reason    IndexFreshnessReason

	// FastPath is true exactly when Reason ==
	// IndexFreshnessReasonStatMismatch — the size comparison alone was
	// conclusive and no content hash was computed. A fresh=true verdict
	// is NEVER reached via the fast path alone: a matching size is
	// treated as inconclusive on its own and always falls through to a
	// full Fingerprint recompute before VerifyIndex will report
	// Fresh=true.
	FastPath bool

	// ModTimeDrift reports that the cohort file's modification time no
	// longer matches the snapshot taken at index build while its size
	// does. On its own that is NOT staleness — mtime resolution is not
	// preserved across filesystems (an S3 LastModified is whole
	// seconds against the index's recorded nanoseconds), so the verdict
	// comes from the fingerprint and this flag is how a caller sees
	// that the cheap stat pair disagreed and the hash overruled it.
	// Always false alongside IndexFreshnessReasonStatMismatch, which is
	// a size verdict.
	ModTimeDrift bool
}

VerifyIndexResult is the outcome of Service.VerifyIndex: whether the sidecar index at IndexPath is still fresh against its source cohort, why, and whether that verdict was reached via the size fast-path alone (FastPath=true) or required a full content-hash recompute (FastPath=false).

type VerifyResult added in v0.8.0 

type VerifyResult struct {
	Errors   []*errors.CodedError       `json:"errors"`
	Warnings []encoding.CohesionWarning `json:"warnings"`

	// SetWidthHeadroom reports, per set_* field in the canonical
	// schema, how much bitmask capacity is left and which rung a widen
	// would promote it to. It is not a verdict — an archive with zero
	// headroom is perfectly valid — it is FORESIGHT: `shard add` widens
	// a set field across every shard payload when the merged dictionary
	// outgrows the rung, and without this the only way to discover the
	// field was one member from its ceiling is to pay for the rewrite.
	// Nil for archives with no set fields.
	SetWidthHeadroom []encoding.SetWidthHeadroom `json:"set_width_headroom,omitempty"`

	// GroupIndexHeadroom reports, per parent group in the canonical
	// schema (format 0x02), how many entries its dictionary holds against
	// its ceiling — the u32 index space for an indexed group, one entry
	// for a constant group. Foresight in the same sense as
	// SetWidthHeadroom: a union past the u32 space is fatal
	// (PULSE_SHARD_DICT_WIDTH_OVERFLOW), and a constant group's next
	// distinct value promotes it archive-wide. Nil for ungrouped archives.
	GroupIndexHeadroom []encoding.GroupIndexHeadroom `json:"group_index_headroom,omitempty"`
}

VerifyResult carries the structured outcome of VerifyShardArchive. Errors aggregates every fatal cohesion failure discovered while walking the archive's shards; Warnings carry non-fatal divergences (per-field description drift, aggregate-record-count drift). An empty Errors slice means the archive is structurally sound.

The shape parallels descriptor.Envelope's {errors, warnings} so callers can lift entries straight into the --json envelope without reshaping. encoding.CohesionWarning is the canonical warning carrier (shared with insert-time cohesion validators).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.