Documentation
¶
Overview ¶
Package datasync implements the `sync` command: a continuous, heterogeneous data sync.
A sync performs an initial copy of the source tables into the target, then streams ongoing changes from the source to the target indefinitely until the context is cancelled. Unlike `move`, there is no checksum and no cutover — the sync simply keeps the target caught up with the source for as long as it runs. Cancelling the command (SIGINT/SIGTERM, or a caller-cancelled context) drains the in-flight backlog and exits cleanly.
The source is either a built-in MySQL binlog client (constructed from SourceDSN) or a caller-injected change.Source — e.g. a Vitess VStream. The target is written through an applier; today that is a single-target MySQLApplier, but the applier abstraction is what makes the sync heterogeneous: a future Postgres applier would let this sync MySQL → Postgres without changing the runner.
The package is deliberately named datasync (not sync) so it does not shadow the standard library's sync package. The CLI command is still `sync` (the kong field name drives the command name, the same way migration.Migration is exposed as `migrate`).
Index ¶
- type Runner
- func (r *Runner) Cancel()
- func (r *Runner) ChecksumReady() <-chan struct{}
- func (r *Runner) ChecksumStats() checksum.LocklessCheckerStats
- func (r *Runner) Close() error
- func (r *Runner) DumpCheckpoint(ctx context.Context) error
- func (r *Runner) FirstCleanPass() <-chan struct{}
- func (r *Runner) InitialChecksumStats() (stats checksum.LocklessCheckerStats, ok bool)
- func (r *Runner) Progress() status.Progress
- func (r *Runner) Run(ctx context.Context) (retErr error)
- func (r *Runner) SetLogger(logger *slog.Logger)
- func (r *Runner) SetMetricsSink(sink metrics.Sink)
- func (r *Runner) Status() string
- func (r *Runner) TargetUnderLoad() bool
- type Sync
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner executes a Sync: an initial copy followed by continuous replication that runs until the context is cancelled.
func NewRunner ¶
NewRunner validates the Sync config and returns a Runner. The CLI supplies defaults via kong; programmatic callers get the same defaults applied here as a safety net.
func (*Runner) Cancel ¶
func (r *Runner) Cancel()
Cancel stops the sync by cancelling the Run context. Safe to call before Run has started (a no-op until Run sets its cancel function).
func (*Runner) ChecksumReady ¶
func (r *Runner) ChecksumReady() <-chan struct{}
ChecksumReady returns a channel that is closed once the lockless checker has been constructed — that is, when the initial copy and the post-copy flush have completed and runChecksum has started.
func (*Runner) ChecksumStats ¶
func (r *Runner) ChecksumStats() checksum.LocklessCheckerStats
ChecksumStats returns a point-in-time snapshot of lockless-checksum counters. Returns the zero value when the checker has not yet been constructed (initial copy still running).
func (*Runner) Close ¶
Close releases resources. Safe to call once after Run returns. The applier's worker lifecycle is owned by the copier (which stops it after the initial copy), so Close does not stop it. An injected change.Source and injected applier/target are owned by the caller, but Close still calls change.Source.Close (documented idempotent) to release the runner's reference.
func (*Runner) DumpCheckpoint ¶
DumpCheckpoint records the copier watermark (and, for continuous sync, the change-feed position) on the target so a restart can resume a partial copy and the stream. It is a no-op until the copy pipeline is built.
func (*Runner) FirstCleanPass ¶
func (r *Runner) FirstCleanPass() <-chan struct{}
FirstCleanPass returns a channel that is closed the first time the lockless checksum completes a clean pass — i.e. every chunk was READ-verified equal (on its initial read or via retry) and no chunk needed a recopy. A pass that repaired chunks via recopy does not qualify; the repaired ranges are re-verified by the following pass before the signal can fire. Programmatic callers that gate on "data is known consistent" (e.g. the import feature) should block on this channel.
The accessor is non-blocking: it returns immediately with a channel the caller can wait on. The Runner-owned channel is closed by an internal goroutine once the checker fires its own FirstCleanPass — so it's safe to call before Run, after Run, or from a watchdog.
func (*Runner) InitialChecksumStats ¶ added in v0.18.0
func (r *Runner) InitialChecksumStats() (stats checksum.LocklessCheckerStats, ok bool)
InitialChecksumStats returns the statistics of the initial verification, the one that repairs, once it has completed; ok is false before then. ChecksumStats reports the run in progress, and continuous verification starts its counters from zero.
func (*Runner) Progress ¶
Progress returns a structured snapshot of the sync's current state: the phase, a short human-readable summary, and per-table copy progress during the initial copy.
func (*Runner) Run ¶
Run performs the initial copy and then streams changes continuously until ctx is cancelled. A clean cancellation during the continuous phase returns nil; a cancellation during setup, the initial copy, or the secondary-index restore can return the wrapped context error. A fatal source event (e.g. DDL) returns an error.
func (*Runner) SetMetricsSink ¶ added in v0.17.0
SetMetricsSink installs the destination for this run's metrics, including the workflow phase transitions reported by status.Tracker. It must be called before Run; a nil sink is ignored.
func (*Runner) Status ¶
Status returns the periodic human-readable report for logging: a header line plus one indented row per subsystem (see status.Block). It does not log itself; status.WatchTask (when used) logs the returned value.
func (*Runner) TargetUnderLoad ¶ added in v0.18.0
TargetUnderLoad is safe before and during Run. Injected change sources can use it as their ClientConfig.UnderLoad callback, matching the built-in feed's adaptive flush control. Synchronous replication writes do not use the applier worker queue, so resizing that queue alone cannot control flushes.
type Sync ¶
type Sync struct {
// Common holds the flags shared with migrate and move: thread counts,
// --max-connections (each source and target pool), --max-commit-latency,
// autoscaling, TLS and --checkpoint-max-age. Sync does not embed
// flags.Cutover: it performs no cutover and takes no table locks.
flags.Common
SourceDSN string `name:"source-dsn" help:"Where to sync the tables from." default:"spirit:spirit@tcp(127.0.0.1:3306)/src"`
TargetDSN string `name:"target-dsn" help:"Where to sync the tables to." default:"spirit:spirit@tcp(127.0.0.1:3306)/dest"`
// FlushInterval controls how often buffered changes are applied to the
// target during continuous sync — i.e. the replication latency vs.
// batching trade-off. Defaults to change.DefaultFlushInterval.
FlushInterval time.Duration `name:"flush-interval" help:"How often to flush buffered changes to the target during continuous sync." default:"30s"`
// DeferSecondaryIndexes creates the target tables without their deferrable regular
// indexes, then adds the indexes back once the initial copy has completed.
// Bulk-loading a table with fewer indexes is faster and lighter on temporary
// space; the indexes are rebuilt in one ALTER per table afterwards. Only
// safe when the target is not yet serving reads, because the tables briefly
// lack their secondary indexes. UNIQUE/FULLTEXT/SPATIAL indexes are kept on
// the initial CREATE, as is a regular index required by AUTO_INCREMENT, the
// same as `move --defer-secondary-indexes`.
DeferSecondaryIndexes bool `` /* 149-byte string literal not displayed */
// Force, when set, makes the runner wipe the sync-owned objects on the
// target at startup — the target copies of the source tables plus the sync
// checkpoint table — *unless* a resumable checkpoint exists, i.e. it only
// wipes when the copy could not have resumed anyway. A resumable run
// (checkpoint present) is left intact and resumes as normal. The wipe is
// per-table, never DROP DATABASE: sync tolerates a target database shared
// with unrelated tables, and --force must not destroy tables the sync
// never owned. Intended for testing/iterating, where a previous partial
// run can leave the target non-empty with no usable checkpoint, otherwise
// tripping the fresh-sync target-empty guard.
Force bool `` /* 159-byte string literal not displayed */
// Source optionally provides a pre-constructed change.Source to use
// for replication instead of constructing a built-in MySQL-binlog
// client from SourceDSN. When set, the runner uses this as the change
// feed. SourceDSN is still required for source-side SQL (SHOW FULL TABLES,
// SHOW CREATE TABLE, the initial-copy SELECTs). Setting Source requires
// setting Applier (see below).
//
// Intended for callers (e.g. a Vitess VStream import) that
// need a non-MySQL-binlog change source.
Source change.Source `kong:"-"`
// Applier optionally provides a pre-constructed applier.Applier. When
// set, the runner uses this instead of constructing a MySQL
// single-target MySQLApplier from the target. Required when Source is set: the
// injected change.Source needs the same applier instance the copier
// uses, so all writes flow through one logical apply path.
Applier applier.Applier `kong:"-"`
// Target optionally supplies a pre-opened target. When nil, a single
// MySQL target is opened from TargetDSN (auto-creating the database if
// it does not exist). Sync writes to exactly one logical target — there
// is no N:M fan-out.
Target *applier.Target `kong:"-"`
}
Sync is the configuration for a continuous data sync. The exported, kong-tagged fields are the CLI surface; the kong:"-" fields are for programmatic callers (e.g. a Vitess VStream import) that inject a non-MySQL change source and/or a custom applier.
func (*Sync) Run ¶
Run is the kong CLI entry point. It runs the sync until the process receives SIGINT/SIGTERM, then drains and exits cleanly. Programmatic callers that want to supply their own context should construct a Runner via NewRunner and call Runner.Run directly.
Signal handling is two-stage: the first SIGINT/SIGTERM cancels the context for a graceful drain, and a second forces an immediate exit. The force-quit matters because the change feed can be slow to unwind (e.g. a busy or repeatedly-reconnecting binlog stream), and without it a second Ctrl+C would be swallowed — making the command feel hung.
Source Files