fleet

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Nov 4, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Overview

Package fleet drives fan-out maintenance across every tenant schema in a relational database: rebinding a fleet onto a new schema-template version, and building the indexes that rebind leaves unbuilt.

The defining property is that a fan-out is NOT one transaction. RFC-204's original migration flow put the whole fleet rebind inside the single catalog transaction that saved the new template ("for each schema bound to tmpl, call RepairSchema(txn, ...)" then "commit catalog transaction"). That cannot work at fleet scale — FDB caps a transaction at 5 s and 10 MB, and a rebind touches one catalog row plus a store-header reconciliation per schema — and it inverts failure isolation: a single poisoned tenant aborts the commit and rolls back every healthy tenant with it.

So atomicity is deliberately forfeited. Each schema gets its OWN transaction, and idempotence replaces atomicity as the correctness argument: a run that dies halfway leaves a well-defined partial state, and re-running converges. The resume key is the schema row's own TEMPLATE_VERSION (listSchemasImpl surfaces it), so a schema already bound at or above the target version is skipped without a write. A fan-out is therefore restartable at any point, and running it twice costs one catalog read per already-done tenant.

Per-target failures are collected, never fatal — the same contract as SweepSPFreshIndexes: one corrupt tenant must not halt fleet maintenance. The pass continues and the joined error reports every failure.

Index

Constants

View Source 
const DefaultConcurrency = 4

DefaultConcurrency is deliberately small: fan-out is background maintenance competing with live tenant traffic on the same cluster, and each slot can hold an OnlineIndexer that is itself already batching aggressively.

Variables

This section is empty.

Functions

func GuardNotCatalog 

func GuardNotCatalog(_ *keyspace.RelationalKeyspace, t Target) error

GuardNotCatalog rejects a target that is, or overlaps, the relational catalog store.

This is reachable, not hypothetical: RecordLayerStoreCatalog.Initialize persists a /__SYS database row and a /__SYS/CATALOG schema row into the catalog it is initialising, so an unfiltered ListSchemas hands the catalog back as an ordinary fan-out target.

Only the system database can overlap the catalog: the catalog store is (NULL, NULL, 0) and every user schema store is three longs (domain, database, schema), so no user schema shares a prefix with it.

func LatestVersions 

func LatestVersions(
	ctx context.Context,
	db *recordlayer.FDBDatabase,
	cat api.StoreCatalog,
	targets []Target,
) (map[string]int, error)

LatestVersions resolves, for each distinct template among targets, the latest version stored in the catalog.

A template with no stored version is an error rather than a silent zero: it means the fleet listing and the template catalog disagree, and rebinding onto "version 0" would either fail per tenant or, worse, pass an assertion no tenant can violate.

func NextTemplateVersion 

func NextTemplateVersion(ctx context.Context, db *recordlayer.FDBDatabase, cat api.StoreCatalog, templateName string) (int, error)

NextTemplateVersion returns one past the latest stored version of templateName, i.e. the version a new template must carry to be accepted. Returns 1 when the template does not exist yet.

func PendingIndexes 

func PendingIndexes(
	ctx context.Context,
	db *recordlayer.FDBDatabase,
	md *recordlayer.RecordMetaData,
	ss subspace.Subspace,
	serializer *recordlayer.TransformedRecordSerializer,
) ([]*recordlayer.Index, error)

PendingIndexes returns the indexes on one store that are not readable — DISABLED or WRITE_ONLY — in deterministic name order.

The states come from the OPENED store's GetAllIndexStates, not from GetAllIndexStatesMap and not from a raw read of the index-state subspace. The three have different domains: GetAllIndexStatesMap returns only the persisted non-readable entries, while GetAllIndexStates walks every index in the metadata and defaults the missing ones to READABLE. An index added by a metadata evolution has NO state key until the store header is reconciled at open, so only opening the store and asking it produces the post- reconciliation truth a build must act on.

serializer is the one the store is opened with (Options.Serializer): the open reconciles the header, which may rebuild a new index over the records inline.

func PinnedMetadata 

func PinnedMetadata(ctx context.Context, db *recordlayer.FDBDatabase, cat api.StoreCatalog, t Target) (*recordlayer.RecordMetaData, error)

PinnedMetadata resolves the record-layer metadata for one schema at the template version the schema row is PINNED to.

Never the latest version: a store pinned to template v1 opened with v2 metadata would have anything honouring checkPossiblyRebuild WRITE to the store from what the caller believes is a read. The pinned version is also simply the truth — it is the metadata the records were written under.

A fan-out index build therefore only ever builds indexes the schema is actually bound to. Making a new index visible to a tenant is the migration step's job, not this one's.

func RestoreTemplateVersion 

func RestoreTemplateVersion(ctx context.Context, db *recordlayer.FDBDatabase, cat *catalog.RecordLayerStoreCatalog,
	ks *keyspace.RelationalKeyspace, templateName string, version int, md []byte,
) error

RestoreTemplateVersion restores a dropped template version from md, its stored MetaData bytes (a backup of the version's catalog row), for the schemas still bound to it: the one way out for a schema whose bound version DROP SCHEMA TEMPLATE removed. ks is the keyspace the deployment opens its schemas with; every bound store's header is read through it. The restore refuses a version that is stored or that no schema binds, a bound store whose header is missing or above md's metadata version, and bytes that are not one history with every stored version of the template (RecordLayerStoreCatalog.RestoreTemplateVersion states the checks; RFC-257 WS-J section 2). Java has no restore.

func SaveTemplate 

func SaveTemplate(ctx context.Context, db *recordlayer.FDBDatabase, cat api.StoreCatalog, tmpl api.SchemaTemplate) (api.SchemaTemplate, error)

SaveTemplate persists tmpl as a new template version in ONE catalog transaction, through the save action (the template catalog's CreateTemplate: the version-monotonicity gate, the relational evolution validator, the version guard, and the carry of a stored template's numbering), and returns the template AS STORED, read back in the same transaction. For a new version of a stored template that differs from tmpl: tmpl holds the numbering its builder gave it, and the stored version keeps the record-type keys, union field numbers and index versions of the stored latest (DIVERGENCES.md "CreateTemplate refuses more than an exact duplicate, and carries a new version"), so a caller must rebind or compare against the returned template.

This is the ONE step of a migration that is legitimately fleet-wide and atomic: it writes a single catalog row, independent of tenant count.

Types

type BuildOptions 

type BuildOptions struct {
	Options
	// Limit is records per transaction.
	Limit int
	// RecordsPerSecond throttles the build.
	RecordsPerSecond int
	// MaxRetries is the per-range retry budget. It also arms the adaptive
	// batch-halving and the records-per-second throttle.
	MaxRetries int
	// TimeLimit stops a build early; the run is resumable.
	TimeLimit time.Duration
	// Logger, when non-nil, receives the indexer's own progress lines.
	Logger *slog.Logger
	// IndexNames, when non-empty, restricts the build to these indexes.
	// Rolling ONE new index across a fleet is the common case, and a
	// tenant that happens to carry an unrelated half-built index must not
	// be dragged into that roll-out.
	//
	// A name that matches nothing on a given tenant is not an error: tenants
	// legitimately sit at different template versions mid-migration, and a
	// fleet job must tolerate that rather than fail the tenant.
	IndexNames []string
}

BuildOptions adds the OnlineIndexer throttle knobs to the shared fan-out options. Zero values mean "leave the indexer's own default alone" — the fleet driver must not silently re-specify a default the indexer owns.

type CatalogTargetError 

type CatalogTargetError struct {
	DatabaseID string
	SchemaName string
	Reason     string
}

CatalogTargetError reports a fan-out target that resolves to the relational catalog itself.

func (*CatalogTargetError) Error 

func (e *CatalogTargetError) Error() string

type Event 

type Event struct {
	Target  Target
	Outcome Outcome
	// Err is set when Outcome is OutcomeFailed or OutcomeRefused.
	Err error
	// Indexes names the indexes acted on (index-build mode only).
	Indexes []string
	// Records is the number of records scanned (index-build mode only).
	Records int64
	// FromVersion / ToVersion bracket a rebind (migration mode only).
	FromVersion int
	ToVersion   int
	// Types is how many record types got a statistic (statistics mode only).
	Types int
}

Event is one progress notification. Exactly one Event is delivered per target per run.

type Options 

type Options struct {
	// Concurrency bounds how many targets are in flight at once. <= 0 means
	// DefaultConcurrency. Each in-flight target holds its own FDB
	// transaction, so this is the real load knob against the cluster.
	Concurrency int
	// Progress, when non-nil, receives one Event per target. Calls are
	// serialised, so the callback does not need to be goroutine-safe.
	Progress func(Event)
	// Serializer is the serializer the targets' user stores are opened with,
	// as the SQL driver opens them (embedded's serializerFromOptions; an
	// encrypting tenant's carries its key manager). Nil opens them without
	// one: every clear or compressed record still reads, and an encrypted one
	// fails the target ("this serializer cannot decrypt") with nothing written.
	Serializer *recordlayer.TransformedRecordSerializer
}

Options are the knobs shared by every fan-out mode.

type Outcome 

type Outcome string

Outcome is what a fan-out did to one target. 

const (
	// OutcomeMigrated means the schema was rebound to a newer template version.
	OutcomeMigrated Outcome = "migrated"
	// OutcomeSkipped means the schema was ALREADY at or above the target
	// version, so no transaction was opened for it. This is the observable
	// that proves idempotent resume actually skipped work rather than
	// silently redoing it.
	OutcomeSkipped Outcome = "skipped"
	// OutcomeBuilt means at least one index was driven to READABLE.
	OutcomeBuilt Outcome = "built"
	// OutcomeNoWork means the store had no DISABLED/WRITE_ONLY index, or every
	// such index was published by another builder before its session ran.
	OutcomeNoWork Outcome = "no-work"
	// OutcomeFailed means this target errored. Other targets still ran.
	OutcomeFailed Outcome = "failed"
	// OutcomeRefused means the catalog guard rejected the target.
	OutcomeRefused Outcome = "refused"
)
const OutcomeCollected Outcome = "collected"

OutcomeCollected means the schema's statistics were collected and stored.

type Result 

type Result struct {
	Total     int
	Migrated  int
	Skipped   int
	Built     int
	Collected int
	NoWork    int
	Failed    int
	Refused   int
	// Failures carries one entry per failed or refused target.
	Failures []*TargetError
}

Result tallies a fan-out.

The per-outcome counts sum to Total on a pass that ran to completion. On a context-cancelled pass they sum to LESS: the shortfall is the targets never attempted, and the returned error carries the context error. Do not treat Total as "targets handled".

func BuildAll 

func BuildAll(
	ctx context.Context,
	db *recordlayer.FDBDatabase,
	cat api.StoreCatalog,
	ks *keyspace.RelationalKeyspace,
	databaseID string,
	opts BuildOptions,
) (Result, error)

BuildAll is the whole index fan-out in one call: enumerate the schemas of databaseID (empty means every database) and build their pending indexes.

func BuildIndexes 

func BuildIndexes(
	ctx context.Context,
	db *recordlayer.FDBDatabase,
	cat api.StoreCatalog,
	ks *keyspace.RelationalKeyspace,
	targets []Target,
	opts BuildOptions,
) (Result, error)

BuildIndexes drives every DISABLED / WRITE_ONLY index on every target to READABLE, one target at a time up to the configured concurrency.

The OnlineIndexer manages its own transactions per range, so this fan-out spans many transactions per target on top of the one-transaction-per-target floor. A poisoned tenant fails alone: its error is recorded against its own target and the remaining tenants keep building.

func CollectAllStatistics 

func CollectAllStatistics(
	ctx context.Context,
	db *recordlayer.FDBDatabase,
	cat api.StoreCatalog,
	ks *keyspace.RelationalKeyspace,
	databaseID string,
	opts StatisticsOptions,
) (Result, error)

CollectAllStatistics collects for every schema in databaseID.

func CollectStatistics 

func CollectStatistics(
	ctx context.Context,
	db *recordlayer.FDBDatabase,
	cat api.StoreCatalog,
	ks *keyspace.RelationalKeyspace,
	targets []Target,
	opts StatisticsOptions,
) (Result, error)

CollectStatistics gathers per-record-type row counts for every target. Exceeding MaxRecordsPerType fails that target without storing partial statistics.

func Migrate 

func Migrate(
	ctx context.Context,
	db *recordlayer.FDBDatabase,
	cat api.StoreCatalog,
	ks *keyspace.RelationalKeyspace,
	targets []Target,
	targetVersion int,
	opts Options,
) (Result, error)

Migrate rebinds every target onto template version targetVersion, ONE TRANSACTION PER SCHEMA.

Targets already at or above targetVersion are skipped without opening a transaction — that skip is the resume mechanism, and it is reported as OutcomeSkipped so a caller can observe that a re-run really did less work rather than silently repeating it.

A rebind that commits without advancing the bound version is treated as a FAILURE, not a success: RepairSchema rebinds to whatever the catalog's latest version happens to be, so if the template save did not land, every tenant would otherwise report "migrated" while still pinned to the old metadata. The landed version is read back inside the same transaction and asserted.

targetVersion is a THRESHOLD, not a destination. RepairSchema has no "rebind to version N" form — it always moves to the catalog's latest — so targetVersion decides only which tenants are already done (skipped) and how far a rebind must have got to count as success. Passing a targetVersion below the latest stored version will rebind tenants PAST it.

func MigrateTemplate 

func MigrateTemplate(
	ctx context.Context,
	db *recordlayer.FDBDatabase,
	cat api.StoreCatalog,
	ks *keyspace.RelationalKeyspace,
	databaseID string,
	tmpl api.SchemaTemplate,
	opts Options,
) (Result, error)

MigrateTemplate is the whole migration in one call: save tmpl as a new version (one catalog transaction), then rebind every schema bound to that template name (one transaction each).

databaseID narrows the fan-out to a single database; empty means every database on the cluster.

The whole call is resumable. A template version that is ALREADY stored is not re-saved: the save is a create, and re-running a migration that died partway through the fan-out would otherwise fail on the template step and never reach the tenants still owed a rebind. Skipping the save and going straight to the fan-out is what makes "run it again" the recovery procedure.

func MigrateToLatest 

func MigrateToLatest(
	ctx context.Context,
	db *recordlayer.FDBDatabase,
	cat api.StoreCatalog,
	ks *keyspace.RelationalKeyspace,
	targets []Target,
	opts Options,
) (Result, error)

MigrateToLatest rebinds every target onto the latest stored version of the template THAT TARGET is bound to.

Version resolution is per TEMPLATE, never fleet-wide, and that is the whole point of this function existing rather than callers computing one number. A database may hold schemas bound to different templates, and those templates advance independently. Judging every tenant against a single fleet-wide number — the maximum across templates, say — breaks every tenant whose own template has not reached it: RepairSchema can only move a schema to ITS OWN template's latest, so the asserted read-back in Migrate correctly rejects the rebind, the tenant is reported as failed, and it never migrates at all. The tenants that need the pass most are exactly the ones it drops.

So the fan-out is executed as one Migrate pass per template, and the tallies are merged.

type StatisticsOptions 

type StatisticsOptions struct {
	Options
	// Collect is passed to the collector for each schema. BatchSize bounds the
	// records per transaction; MaxRecordsPerType caps the work spent on one
	// type: crossing it ABORTS that schema's collection and stores nothing,
	// which surfaces as a per-target failure rather than an OutcomeCollected.
	Collect recordlayer.CollectOptions
}

StatisticsOptions tunes a statistics fan-out.

type Target 

type Target struct {
	DatabaseID      string
	SchemaName      string
	TemplateName    string
	TemplateVersion int
}

Target identifies one tenant schema, as the catalog currently records it. TemplateVersion is the version the schema row is PINNED to — the resume key.

func FilterByTemplate 

func FilterByTemplate(targets []Target, templateName string) []Target

FilterByTemplate narrows targets to those bound to templateName. An empty templateName returns targets unchanged.

func ListTargets 

func ListTargets(ctx context.Context, db *recordlayer.FDBDatabase, cat api.StoreCatalog, databaseID string) ([]Target, error)

ListTargets enumerates the schemas a fan-out would touch. An empty databaseID enumerates every database on the cluster; otherwise the listing is narrowed to that database URI.

Catalog-owned rows are NOT filtered here — enumeration reports what the catalog holds, and GuardNotCatalog decides what may be written. Filtering early would hide the fact that the catalog self-registers.

func (Target) String 

func (t Target) String() string

String renders the target as "database/schema".

type TargetError 

type TargetError struct {
	Target Target
	Err    error
}

TargetError attaches a target to the error it produced, so a joined fleet error still says WHICH tenant broke.

func (*TargetError) Error 

func (e *TargetError) Error() string

func (*TargetError) Unwrap 

func (e *TargetError) Unwrap() error

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.