runtime/me.ahoo.wow.runtime.snapshot/SnapshotMaintenance

SnapshotMaintenance

class SnapshotMaintenance(val context: String, group: String, node: String, models: Collection<AggregateModel<*>>, val targets: SnapshotTargets, kernel: Kernel, codec: EventCodec, log: EventLog, leases: LeaseStore, progress: List<LogProgress>, options: SnapshotOptions = SnapshotOptions(), address: () -> String = { node }, observer: Observer = Observer.NONE, gate: ReplayGate = ReplayGate.OPEN) : SnapshotStatus

Keeps the snapshot targets of one bounded context in step with the log and the code, on every node, every SnapshotOptions.interval:

State revision claims: each node is a member of its state revision of each snapshotted type (in the lease group <group>#snapshot/<aggregate>@<revision>), and claims each target's table (SnapshotRebuildStore.claim) at its revision: a higher revision at once (a new deployment, the start of a rolling upgrade), a lower one only once no member of the claimed revision is left (a rollback). Its snapshot subscriptions write at its revision whatever the claim; the store keeps a node of a lower revision from overwriting the higher one and marks those snapshots stale.
In-place rebuild: a node whose revision a table is claimed at, and which holds the table's rebuild lease (the lease group <group>#snapshot-rebuild/<store>/<aggregate>, so one node rebuilds), rewrites its pending snapshots (SnapshotRebuildStore.pending) from the log, page by page, at their latest version, each run going on from where the last one stopped, and marks the table rebuilt once a scan from its first snapshot finds none left. A node of another revision marks the table rebuilding on each run, so the stale marks its refused saves leave are rewritten.
Backfill: a target that has not caught up with the log since it was first claimed is filled from the beginning of the log, by its store's subscription when that one has no checkpoint yet or is snapshot (the default store's, which wrote every type from the beginning before targets existed), else by a subscription of its own (prepare, <store subscription>#<aggregate>) that stops writing once the target caught up. A target caught up once its subscription's checkpoint in every partition reached the log's head of the previous run; the table is then marked backfilled.
Rebuild requests: the built-in subscription REBUILD_SUBSCRIPTION rewrites, from the log, the snapshots each wow.SnapshotRebuildRequested of the rebuild aggregate names (SnapshotRebuildModel), in each target: in place on a node of the claimed revision, else marked stale (or written at its own revision when missing) for the rebuild.

It answers rebuilding for a type while the table of its primary target is rebuilding or not backfilled.

The claims, rebuilds and backfill checks need a target store's rebuild control plane (SnapshotStore.rebuilds): a store without one is left out of them, with a warning when the node prepares and a count of the fallback (the observer's storeFellBack), and so is its share of a rebuild request.

Constructors

SnapshotMaintenance

constructor(context: String, group: String, node: String, models: Collection<AggregateModel<*>>, targets: SnapshotTargets, kernel: Kernel, codec: EventCodec, log: EventLog, leases: LeaseStore, progress: List<LogProgress>, options: SnapshotOptions = SnapshotOptions(), address: () -> String = { node }, observer: Observer = Observer.NONE, gate: ReplayGate = ReplayGate.OPEN)

Types

Companion

@InternalWowApi

object Companion

Properties

context

val context: String

rebuildSubscription

val rebuildSubscription: Subscription

The built-in subscription that carries out the rebuild requests of the context.

targets

val targets: SnapshotTargets

types

val types: Set<AggregateType>

The snapshotted aggregate types of the context.

Functions

maintain

suspend fun maintain()

One run: claims, rebuilds, backfill checks; see the class.

prepare

suspend fun prepare(): List<SnapshotSubscription>

Claims every target's table at this node's revisions, and gives the subscriptions that backfill the targets whose store subscription is already past the beginning of the log: register them before the node starts.

rebuild

suspend fun rebuild(ref: AggregateRef)

Rebuilds the snapshot of ref from the log in each target of its type (a rebuild request): in place where the table is claimed at this node's revision; elsewhere marked stale, or written at this node's revision when there is none, for the node of the claimed revision to rewrite.

suspend fun rebuild(ref: AggregateRef, streams: List<EventStreamRecord>, state: CachedState)

Rebuilds the snapshot of ref as rebuild does, from streams, its whole log, already replayed as state: the access walk's, which read them to stamp them.

rebuilding

open override fun rebuilding(type: AggregateType): Boolean

start

fun start(scope: CoroutineScope)

Runs maintain every interval in scope, the first one an interval after prepare, until stop.

stop

suspend fun stop()

Stops the runs and leaves this node's revision memberships, so a rollback need not wait for them to expire. Call it once the node's subscriptions stopped: a table claimed at another revision than this node's is marked rebuilding once more, for the stale marks its last refused saves left.

tables

fun tables(): Map<Pair<String, AggregateType>, SnapshotTable>

The table records last seen, by store and aggregate type.