follow

package
v0.2.0 Latest Latest
Warning

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

Go to latest
Published: Nov 6, 2026 License: BSD-2-Clause Imports: 13 Imported by: 0

Documentation

Overview

Package follow tails Tarantool journal files as they are written. It adds two live-read modes on top of the forward-only reader:

  • FollowFile / FollowFileTx — tail a single .xlog, blocking for rows the writer appends, until the file is finalised (its EOF marker appears) or the context is cancelled.
  • FollowDir / FollowDirTx — tail a whole WAL directory: read the current file, and when it is finalised switch to the next file in the rotation chain (validating chain continuity), continuing indefinitely.

Both are exposed as Go 1.23 iterators (iter.Seq2), matching reader.Rows / reader.Txs. Cancellation via context.Context ends the iterator cleanly (without yielding an error), the same way io.EOF ends the reader iterators.

The mechanism is offset-tracked re-open: each poll re-opens the file at the byte offset a prior pass stopped at (reader.OpenAt + reader.Offset) and reads the newly-appended blocks, so already-consumed blocks are never re-read. A partial trailing block or a still-missing EOF marker means "wait for more"; the real EOF marker (reader.SawEOFMarker) means "finalised". Directory and file growth are observed by a pluggable Watcher; the default polls.

Index

Examples

Constants

This section is empty.

Variables

View Source
var (
	// ErrChainBroken is returned when the next file in a directory does not
	// chain onto the current one — its PrevVClock does not equal the current
	// file's VClock. The rotation chain has a gap or has diverged; the follower
	// stops rather than silently skipping records.
	ErrChainBroken = errors.New("follow: chain continuity broken: successor PrevVClock != current VClock")

	// ErrNoStart is returned by FollowDir / FollowDirTx when no start option
	// was given. A directory follow has no default position — pass WithFromHead,
	// WithStartLSN, or WithStartVClock.
	ErrNoStart = errors.New("follow: directory follow requires a start option (WithFromHead/WithStartLSN/WithStartVClock)")
)

Sentinel errors. Callers match with errors.Is.

Functions

func Dir

func Dir(ctx context.Context, dirPath string, ft format.Filetype, opts ...Option) iter.Seq2[format.XRow, error]

Dir tails a WAL directory across rotation: it reads the start file (set by WithFromHead / WithStartLSN / WithStartVClock — required, else ErrNoStart), and when that file is finalised it switches to the next file in the chain and continues, indefinitely, until ctx is cancelled. Each switch validates chain continuity (the successor's PrevVClock must equal the current file's VClock); a break ends the iterator with ErrChainBroken.

for row, err := range follow.Dir(ctx, walDir, format.FiletypeXLOG, follow.WithFromHead()) {
    if err != nil { return err }
    // ... process row ...
}

By default only finalised files are read; WithReadInprogress also tails the active .inprogress file for lower latency.

Example

ExampleDir tails a whole WAL directory, following rotation: it reads the current file and, when that file is finalised, switches to the next file in the chain — indefinitely, until the context is cancelled. A directory follow must be given a start position (here, the head of the chain). This example breaks out after the two rows it knows are present; a real consumer keeps ranging until it cancels the context.

dir, err := os.MkdirTemp("", "follow-dir")
if err != nil {
	log.Fatal(err)
}

// Two finalised files forming a chain: file 2's PrevVClock == file 1's VClock.
writeExampleChainFile(dir, format.VClock{1: 1}, nil, 1)
writeExampleChainFile(dir, format.VClock{1: 2}, format.VClock{1: 1}, 2)

ctx, cancel := context.WithCancel(context.Background())

seq := follow.Dir(ctx, dir, format.FiletypeXLOG, follow.WithFromHead())

n := 0

for r, err := range seq {
	if err != nil {
		log.Fatal(err)
	}

	fmt.Printf("replica %d lsn %d\n", r.ReplicaID, r.LSN)

	if n++; n == 2 {
		break // stop following (a real consumer would cancel on its own condition)
	}
}

cancel()

_ = os.RemoveAll(dir)
Output:
replica 1 lsn 1
replica 1 lsn 2

func DirTx

func DirTx(ctx context.Context, dirPath string, ft format.Filetype, opts ...Option) iter.Seq2[*reader.Transaction, error]

DirTx is Dir at logical-transaction granularity, yielding one *reader.Transaction per committed tx across the whole chain.

func File

func File(ctx context.Context, path string, opts ...Option) iter.Seq2[format.XRow, error]

File tails a single .xlog (or snap/vylog) file, yielding each row the writer appends. It blocks for new rows until the file is finalised — its EOF marker is written — at which point the iterator ends, or until ctx is cancelled (also a clean end). A terminal error (corruption, I/O) is yielded once as the last pair.

for row, err := range follow.File(ctx, path) {
    if err != nil { return err }
    // ... process row ...
}

Use WithStartOffset to resume from a byte offset captured earlier via reader.Offset / Follower.Offset.

Example

ExampleFile tails a single .xlog file. It blocks for rows as they are appended and ends when the file is finalised (its EOF marker is written). Here the file is already complete, so the follow drains it and stops.

dir, err := os.MkdirTemp("", "follow-file")
if err != nil {
	log.Fatal(err)
}

path := filepath.Join(dir, "00000000000000000000.xlog")
writeExampleFile(path, 1, 2, 3)

// Tail it: the loop yields each row, then ends at the EOF marker.
for r, err := range follow.File(context.Background(), path) {
	if err != nil {
		log.Fatal(err)
	}

	fmt.Println(r.LSN)
}

_ = os.RemoveAll(dir)
Output:
1
2
3

func FileTx

func FileTx(ctx context.Context, path string, opts ...Option) iter.Seq2[*reader.Transaction, error]

FileTx is File at logical-transaction granularity: it yields one *reader.Transaction per committed tx, waiting for the commit row of an incomplete tx at the tail before emitting it.

Types

type Follower

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

Follower is the stateful driver behind the iterators, for callers that want to pull rows one at a time and checkpoint progress across restarts via Offset. The iterator functions (File / Dir) are the simpler surface; reach for Follower when you need the resume offset.

A Follower is not safe for concurrent use.

Example

ExampleFollower shows the pull-style driver with offset checkpointing. Persist Offset() (with the current file) to resume after a restart via WithStartOffset.

dir, err := os.MkdirTemp("", "follow-er")
if err != nil {
	log.Fatal(err)
}

path := filepath.Join(dir, "00000000000000000000.xlog")
writeExampleFile(path, 1, 2)

f := follow.NewFileFollower(path)
ctx := context.Background()

for {
	r, err := f.Next(ctx)
	if errors.Is(err, io.EOF) {
		break // the file was finalised
	}

	if err != nil {
		log.Fatal(err)
	}

	fmt.Println(r.LSN)

	_ = f.Offset() // resume point: follow.WithStartOffset(off) on a later run
}

_ = f.Close()
_ = os.RemoveAll(dir)
Output:
1
2

func NewDirFollower

func NewDirFollower(dirPath string, ft format.Filetype, opts ...Option) *Follower

NewDirFollower returns a Follower tailing a rotation chain. See Dir; a start option is required (else Next returns ErrNoStart). The start file is resolved lazily on the first Next so it can honour ctx.

func NewFileFollower

func NewFileFollower(path string, opts ...Option) *Follower

NewFileFollower returns a Follower tailing a single file. See File for the semantics; WithStartOffset sets the initial resume point.

func (*Follower) Close

func (f *Follower) Close() error

Close releases the underlying open file, if any.

func (*Follower) Next

func (f *Follower) Next(ctx context.Context) (format.XRow, error)

Next returns the next row, blocking (subject to ctx) for appended rows or a rotated file. It returns io.EOF when a single-file follow reaches the finalised end, and ctx.Err() on cancellation. A directory follow never returns io.EOF.

func (*Follower) Offset

func (f *Follower) Offset() int64

Offset returns the byte offset within the current file from which a fresh follow would resume without skipping rows — a clean block boundary. Persist it (alongside the current file path, via the dir chain) to resume after a restart with WithStartOffset.

type Option

type Option func(*config)

Option configures a follower. Options are applied in order.

func WithFromHead

func WithFromHead() Option

WithFromHead starts a directory follow at the lowest-signature file — i.e. replay the whole chain on disk, then follow live.

func WithPollInterval

func WithPollInterval(d time.Duration) Option

WithPollInterval sets the cadence the default Watcher re-checks for new bytes or files. Default 200ms. Ignored when a custom Watcher is supplied.

func WithReadInprogress

func WithReadInprogress() Option

WithReadInprogress lets a directory follow tail the active <sig>.<ext>.inprogress file directly for lowest latency, instead of waiting for it to be finalised (renamed) before reading. Off by default: only finalised, index-visible files are read, which keeps chain validation simple and avoids the rename race.

func WithReaderOptions

func WithReaderOptions(opts ...reader.Option) Option

WithReaderOptions threads reader Options (e.g. SkipCorruptTx, AcceptV012, WithAliasBodies) into every underlying open. IgnoreMissingEOF is always applied internally — a follower must tolerate the not-yet-finalised tail.

func WithStartLSN

func WithStartLSN(replicaID uint32, lsn int64) Option

WithStartLSN starts a directory follow at the file containing (replicaID, lsn) via dir.LocateLSN.

func WithStartOffset

func WithStartOffset(off int64) Option

WithStartOffset begins a single-file follow at a byte offset previously reported by reader.Offset / Follower.Offset (always a clean block boundary). It has no effect on a directory follow. Default 0 (start after the header).

func WithStartVClock

func WithStartVClock(target format.VClock) Option

WithStartVClock starts a directory follow at the file containing target via dir.LocateVClock.

func WithWatcher

func WithWatcher(w Watcher) Option

WithWatcher overrides the default polling Watcher (e.g. with an fsnotify-based one). The follower's correctness does not depend on the Watcher reporting real changes — it re-reads regardless — so a Watcher only affects latency and CPU, never what is emitted.

type Watcher

type Watcher interface {
	Wait(ctx context.Context, path string) error
}

Watcher decides when a follower re-checks the filesystem. Wait blocks until path may have changed (a file grew, a directory gained a file) or ctx is done, in which case it returns ctx.Err(). It is only a hint to re-check, not a guarantee of change — the follower re-reads and re-resolves regardless — so the default pollWatcher simply waits one poll interval. The seam lets an event-driven implementation (e.g. fsnotify) be slotted in via WithWatcher without changing the follower.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL