Documentation
¶
Overview ¶
Package writer implements a single-file, write-once cursor that produces Tarantool xlog / snap / vylog / run / index files in the durable journal format. It is built on top of the pure-byte `format` package and adds the I/O concerns the format package deliberately stays out of.
A Writer:
- opens `<path>.inprogress` exclusively on construction;
- encodes the meta header via format.EncodeMeta;
- accepts rows in one of two API shapes — append-then-commit (WriteRow / CommitTx) or whole-tx (WriteTx);
- flushes per-tx via format.EncodeTxBlock (zstd-compresses payloads at or above CompressThreshold);
- on Close, writes the EOFMarker, flushes, fsyncs per cfg, then atomically renames `<path>.inprogress` → `<path>`.
Strictness defaults: O_EXCL on .inprogress (no silent overwrite), post-Close calls return ErrClosed, mid-tx WriteTx errors out rather than silently flushing.
Index ¶
- Variables
- type BatchOptions
- type BatchWriter
- type Option
- type SyncMode
- type Writer
- func (w *Writer) Close() error
- func (w *Writer) CommitTx() error
- func (w *Writer) Discard() error
- func (w *Writer) Sync() error
- func (w *Writer) WriteBlock(rows []format.XRow) error
- func (w *Writer) WriteRawBlock(block []byte) error
- func (w *Writer) WriteRow(r format.XRow) error
- func (w *Writer) WriteTx(rows []format.XRow) error
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrClosed is returned by any Writer method called after Close or Discard // has finished. ErrClosed = errors.New("writer: closed") // ErrNilDst is returned by NewWriter when dst is nil. ErrNilDst = errors.New("writer: NewWriter: nil dst") // ErrNilMeta is returned by Create / NewWriter when meta is nil. ErrNilMeta = errors.New("writer: nil meta") // ErrPendingWriteRow is returned by WriteTx when the WriteRow accumulator // is non-empty; the caller must CommitTx first. ErrPendingWriteRow = errors.New("writer: WriteTx with pending WriteRow buffer; call CommitTx first") // ErrEmptyRows is returned by WriteTx when given an empty rows slice, and // by WriteBlock / BatchWriter.WriteTx when a transaction has no rows. ErrEmptyRows = errors.New("writer: WriteTx: empty rows") // ErrEmptyBlock is returned by WriteBlock when given an empty txs slice. ErrEmptyBlock = errors.New("writer: WriteBlock: empty txs") // ErrInvalidRawBlock is returned by WriteRawBlock when the bytes are not a // well-formed framed tx block: shorter than a fixheader, or not prefixed by // RowMarker / ZRowMarker. ErrInvalidRawBlock = errors.New("writer: invalid raw block") )
Sentinel errors returned by the Writer. Their identity matters for callers using errors.Is.
Functions ¶
This section is empty.
Types ¶
type BatchOptions ¶
BatchOptions configures a BatchWriter's block-flush policy. A buffered block is flushed when EITHER limit is reached, whichever trips first:
- MaxTxs: flush once this many logical transactions are buffered.
- MaxBytes: flush once the buffered payload reaches this many bytes.
A zero limit disables that dimension. With both zero the BatchWriter never auto-flushes — transactions accumulate into one block until Flush or Close, which is rarely what you want for a large stream. MaxBytes is measured as the encoded (pre-compression) payload size buffered so far — the exact plain block size — so it is a tight target for on-disk block size before zstd.
type BatchWriter ¶
type BatchWriter struct {
// contains filtered or unexported fields
}
BatchWriter packs many independent logical transactions into compressed tx blocks. Each WriteTx call is one logical transaction: BatchWriter assigns its TSN/commit flags (via assignTxIDs) and appends its rows to a flat buffer. When a BatchOptions threshold is reached the buffer is flushed as one physical block via the writer's internal block framing — so each buffered transaction keeps its own identity (single-row txs stay autocommit) while still sharing a zstd-compressed block. This is the "caching dumper" shape: the grouping Tarantool's xlog uses, which plain WriteTx cannot produce because it merges its rows into one logical tx.
Not safe for concurrent use. The BatchWriter takes ownership of the underlying Writer for its lifetime; do not call the Writer's WriteRow / CommitTx / WriteTx / WriteBlock directly while a BatchWriter wraps it.
WriteTx encodes each transaction into the pending byte buffer immediately and never retains the caller's rows, so rows fed straight from a reader's recycled decode buffers (including aliased BodyRaw) need no cloning.
Example ¶
ExampleBatchWriter packs many independent single-row transactions into compressed blocks. Each transaction keeps its own identity (it stays a single-row autocommit tx) while sharing a physical block, which is the shape Tarantool's own xlog uses.
package main
import (
"bytes"
"fmt"
"github.com/google/uuid"
"github.com/tarantool/go-iproto"
"github.com/vmihailenco/msgpack/v5"
"github.com/tarantool/go-xlog/format"
"github.com/tarantool/go-xlog/reader"
"github.com/tarantool/go-xlog/writer"
)
func exampleMeta() *format.Meta {
return &format.Meta{
Filetype: format.FiletypeXLOG,
Version: "example",
InstanceUUID: uuid.MustParse("11111111-2222-3333-4444-555555555555"),
}
}
// exampleBody msgpack-encodes a one-element tuple so the row body is a valid
// msgpack value the reader can round-trip.
func exampleBody(v int) []byte {
b, err := msgpack.Marshal([]int{v})
if err != nil {
panic(err)
}
return b
}
func main() {
var buf bytes.Buffer
w, err := writer.NewWriter(&buf, exampleMeta())
if err != nil {
panic(err)
}
// Flush a block every two transactions (a real caller would use a larger
// MaxBytes so blocks cross the zstd compression threshold).
bw := writer.NewBatchWriter(w, writer.BatchOptions{MaxTxs: 2})
for lsn := int64(1); lsn <= 5; lsn++ {
if err := bw.WriteTx([]format.XRow{
{Type: iproto.IPROTO_INSERT, LSN: lsn, BodyRaw: exampleBody(int(lsn))},
}); err != nil {
panic(err)
}
}
if err := bw.Close(); err != nil { // Flushes the trailing block + closes.
panic(err)
}
r, err := reader.NewReader(bytes.NewReader(buf.Bytes()))
if err != nil {
panic(err)
}
defer func() { _ = r.Close() }()
txs, multiRow := 0, 0
for tx, err := range r.Txs() {
if err != nil {
panic(err)
}
txs++
if len(tx.Rows) != 1 {
multiRow++
}
}
fmt.Printf("%d transactions preserved, %d merged\n", txs, multiRow)
}
Output: 5 transactions preserved, 0 merged
Example (Pipeline) ¶
ExampleBatchWriter_pipeline streams rows straight from a reader into a BatchWriter with no cloning — the caching-dumper repack path. Because BatchWriter encodes each transaction eagerly into its pending buffer, it never retains the caller's rows, so rows whose BodyRaw aliases the reader's recycled read buffer (reader.WithAliasBodies, the fastest read mode) can be forwarded directly.
package main
import (
"bytes"
"fmt"
"github.com/google/uuid"
"github.com/tarantool/go-iproto"
"github.com/vmihailenco/msgpack/v5"
"github.com/tarantool/go-xlog/format"
"github.com/tarantool/go-xlog/reader"
"github.com/tarantool/go-xlog/writer"
)
func exampleMeta() *format.Meta {
return &format.Meta{
Filetype: format.FiletypeXLOG,
Version: "example",
InstanceUUID: uuid.MustParse("11111111-2222-3333-4444-555555555555"),
}
}
// exampleBody msgpack-encodes a one-element tuple so the row body is a valid
// msgpack value the reader can round-trip.
func exampleBody(v int) []byte {
b, err := msgpack.Marshal([]int{v})
if err != nil {
panic(err)
}
return b
}
func main() {
// Build a small source xlog to repack.
var srcBuf bytes.Buffer
sw, err := writer.NewWriter(&srcBuf, exampleMeta())
if err != nil {
panic(err)
}
for lsn := int64(1); lsn <= 5; lsn++ {
if err := sw.WriteTx([]format.XRow{
{Type: iproto.IPROTO_INSERT, LSN: lsn, BodyRaw: exampleBody(int(lsn))},
}); err != nil {
panic(err)
}
}
if err := sw.Close(); err != nil {
panic(err)
}
// Repack: read with aliased bodies, forward each row to the BatchWriter
// without cloning.
rd, err := reader.NewReader(bytes.NewReader(srcBuf.Bytes()), reader.WithAliasBodies())
if err != nil {
panic(err)
}
defer func() { _ = rd.Close() }()
var dstBuf bytes.Buffer
dw, err := writer.NewWriter(&dstBuf, exampleMeta())
if err != nil {
panic(err)
}
bw := writer.NewBatchWriter(dw, writer.BatchOptions{MaxTxs: 2})
for rd.Scan() {
row := rd.Row() // BodyRaw aliases the read buffer — no clone needed.
if err := bw.WriteTx([]format.XRow{row}); err != nil {
panic(err)
}
}
if err := rd.Err(); err != nil {
panic(err)
}
if err := bw.Close(); err != nil {
panic(err)
}
// Read the repacked output back to confirm every record survived.
out, err := reader.NewReader(bytes.NewReader(dstBuf.Bytes()))
if err != nil {
panic(err)
}
defer func() { _ = out.Close() }()
count := 0
for _, err := range out.Rows() {
if err != nil {
panic(err)
}
count++
}
fmt.Printf("repacked %d records (no clone)\n", count)
}
Output: repacked 5 records (no clone)
func NewBatchWriter ¶
func NewBatchWriter(w *Writer, opt BatchOptions) *BatchWriter
NewBatchWriter returns a BatchWriter that flushes blocks to w according to opt. It does not take any action on its own; call WriteTx to feed transactions and Close (or Flush) to emit the final block.
func (*BatchWriter) Close ¶
func (b *BatchWriter) Close() error
Close flushes any buffered transactions, then closes the underlying Writer (writing the EOF marker and, for a file writer, renaming into place). If the final flush fails the Writer is left open so the caller can Discard.
func (*BatchWriter) Flush ¶
func (b *BatchWriter) Flush() error
Flush writes any buffered transactions as one block and resets the buffer. It is a no-op when nothing is buffered. On a write error the buffer is left intact so the caller may retry or Discard.
func (*BatchWriter) WriteTx ¶
func (b *BatchWriter) WriteTx(rows []format.XRow) error
WriteTx buffers one logical transaction (rows sharing a tx). AssignTxIDs is applied to rows (shared TSN = first row's LSN, only the last row commits), mutating them in place, then the rows are *encoded* straight into the pending payload buffer — the caller's rows are not retained, so they may be reused or recycled immediately. When a flush threshold is reached the buffer is written as a single (possibly compressed) block. Returns ErrEmptyRows if rows is empty.
type Option ¶
type Option func(*writerCfg)
Option configures a Writer at construction time. Options are applied in order; the Writer is otherwise immutable.
func NoCompression ¶
func NoCompression() Option
NoCompression disables zstd compression for every tx written by this writer, regardless of payload size — sugar for WithCompression(format.Compression{Disabled: true}). Useful for test fixtures whose byte content needs to be inspected, and for tools whose downstream consumer cannot decompress.
func Sync ¶
Sync chooses the Close-time durability behaviour: SyncNone (skip), SyncNormal (f.Sync(), the default), or SyncDataSync (Fdatasync on Linux, fallback to f.Sync() elsewhere).
func Version ¶
Version sets the Meta.Version string written to the file's text header when the caller-supplied Meta has Version == "". Pass-through Meta.Version values always win — this option only fills in the blank.
func WithCompression ¶
func WithCompression(c format.Compression) Option
WithCompression sets the block compression policy for the writer: the zstd level, the minimum payload size at which a block is compressed, or disabling compression entirely. A zero-value field takes its default (ZstdLevel / CompressThreshold), so format.Compression{Level: 9} tunes only the level.
type SyncMode ¶
type SyncMode int
SyncMode controls the Close-time durability behaviour of a Writer. Only Close-time sync is exposed in v1 — per-write syncs are not.
const ( // SyncNone disables Close-time fsync. The OS may delay the bytes hitting // the platter past Close return; only Rename atomicity remains. SyncNone SyncMode = iota // SyncNormal is the default: f.Sync() on Close before rename. SyncNormal // SyncDataSync uses syscall.Fdatasync on Linux (data-only sync; skips // the inode metadata flush f.Sync() implies). Falls back to f.Sync() // on platforms without Fdatasync. SyncDataSync )
type Writer ¶
type Writer struct {
// contains filtered or unexported fields
}
Writer is a single-file, write-once cursor that produces a Tarantool journal file in the format described by the `format` package. Not safe for concurrent use; create one per writing goroutine.
Lifecycle:
- Create opens `<path>.inprogress` exclusively and writes the meta header.
- WriteRow / CommitTx (or WriteTx) append tx blocks to the file.
- Close writes the EOFMarker, flushes, syncs per cfg, and atomically renames `.inprogress` → final. After Close all methods return ErrClosed.
- Discard is the escape hatch: closes the fd and removes the `.inprogress` file without ever producing a final file. After Discard all methods return ErrClosed.
Example ¶
ExampleWriter writes a couple of single-statement transactions to an in-memory xlog, then reads them back.
package main
import (
"bytes"
"fmt"
"github.com/google/uuid"
"github.com/tarantool/go-iproto"
"github.com/vmihailenco/msgpack/v5"
"github.com/tarantool/go-xlog/format"
"github.com/tarantool/go-xlog/reader"
"github.com/tarantool/go-xlog/writer"
)
func exampleMeta() *format.Meta {
return &format.Meta{
Filetype: format.FiletypeXLOG,
Version: "example",
InstanceUUID: uuid.MustParse("11111111-2222-3333-4444-555555555555"),
}
}
// exampleBody msgpack-encodes a one-element tuple so the row body is a valid
// msgpack value the reader can round-trip.
func exampleBody(v int) []byte {
b, err := msgpack.Marshal([]int{v})
if err != nil {
panic(err)
}
return b
}
func main() {
var buf bytes.Buffer
w, err := writer.NewWriter(&buf, exampleMeta())
if err != nil {
panic(err)
}
for lsn := int64(1); lsn <= 2; lsn++ {
if err := w.WriteTx([]format.XRow{
{Type: iproto.IPROTO_INSERT, LSN: lsn, BodyRaw: exampleBody(int(lsn))},
}); err != nil {
panic(err)
}
}
if err := w.Close(); err != nil {
panic(err)
}
r, err := reader.NewReader(bytes.NewReader(buf.Bytes()))
if err != nil {
panic(err)
}
defer func() { _ = r.Close() }()
count := 0
for _, err := range r.Rows() {
if err != nil {
panic(err)
}
count++
}
fmt.Printf("read back %d record(s)\n", count)
}
Output: read back 2 record(s)
func Create ¶
Create opens path.inprogress for exclusive writing, writes the meta header, and returns a Writer ready for tx submission.
Failure modes:
- the .inprogress file already exists (O_EXCL): error wrapping the underlying os.OpenFile failure (no silent overwrite).
- meta encoding fails: file is closed and removed; the partial .inprogress is not left behind.
func NewWriter ¶
NewWriter returns a Writer that streams to an arbitrary io.Writer instead of a file. It writes the meta header immediately, then accepts tx submissions exactly like a Create'd writer. The file-only concerns do not apply: there is no .inprogress / atomic rename (the caller owns dst), and Close performs no fsync — it flushes the EOF marker and the buffered tail, nothing more. SyncMode and Sync() are no-ops for an in-memory writer.
Useful for producing an xlog into a buffer, a network stream, or a compressing writer, and for fast in-memory round-trip testing.
func (*Writer) Close ¶
Close finalises the file:
- Flushes any pending WriteRow accumulator via CommitTx.
- Writes the 4-byte EOFMarker.
- Flushes the bufio buffer.
- fsyncs per cfg.sync.
- Closes the fd.
- Atomically renames .inprogress → final.
- fsyncs the parent directory so the rename is durable (unless SyncNone).
After Close all further methods return ErrClosed. If any step fails before the rename, the .inprogress file is left in place (the caller can inspect it). The Writer transitions to closed in either outcome — callers must Discard if they want the .inprogress removed after a failed Close.
func (*Writer) CommitTx ¶
CommitTx applies assignTxIDs to the accumulated WriteRow buffer, encodes the rows as one tx block, writes them through the buffered file, and clears the accumulator. No-op when the accumulator is empty.
func (*Writer) Discard ¶
Discard releases the file handle and removes the .inprogress file without promoting it to its final name. Use after a caller-side error that makes the in-flight file unusable.
After Discard all further methods return ErrClosed. Discard is a no-op after Close (returns ErrClosed) because Close has already removed the .inprogress (via rename).
func (*Writer) Sync ¶
Sync flushes the bufio buffer to the kernel and then either f.Sync() or fdatasync() depending on cfg.sync. SyncNone is treated as bufio flush only (no kernel sync). Useful between WriteTx calls for callers that want stronger durability than Close-time sync.
func (*Writer) WriteBlock ¶
WriteBlock writes rows as a single physical tx block, *verbatim* — it does NOT touch TSN or commit flags. The rows' own TSN/commit flags delimit the logical transactions inside the block, so one block may hold many transactions (the shape Tarantool's own xlog uses: e.g. ~50 single-row autocommit txs per zstd block). The payload is zstd-compressed when it crosses CompressThreshold.
Because it trusts the flags, the rows must already carry valid TSN/commit — which is exactly how rows arrive from the reader, making WriteBlock the efficient primitive for copy/repack/truncate. To synthesise ONE transaction from freshly-built rows (LSNs only), use WriteTx, which assigns the TSN and commit flag for you; to pack many independent transactions into blocks without managing flags yourself, use BatchWriter.
WriteBlock must not be interleaved with a pending WriteRow accumulator: it returns ErrPendingWriteRow if that buffer is non-empty, ErrEmptyBlock if rows is empty, and ErrClosed after Close.
Example ¶
ExampleWriter_WriteBlock copies records from one xlog into a single verbatim block of another. WriteBlock writes the rows as-is — it trusts the TSN/commit flags they already carry (here, from the source reader) — so it is the efficient primitive for copy/repack.
package main
import (
"bytes"
"fmt"
"github.com/google/uuid"
"github.com/tarantool/go-iproto"
"github.com/vmihailenco/msgpack/v5"
"github.com/tarantool/go-xlog/format"
"github.com/tarantool/go-xlog/reader"
"github.com/tarantool/go-xlog/writer"
)
func exampleMeta() *format.Meta {
return &format.Meta{
Filetype: format.FiletypeXLOG,
Version: "example",
InstanceUUID: uuid.MustParse("11111111-2222-3333-4444-555555555555"),
}
}
// exampleBody msgpack-encodes a one-element tuple so the row body is a valid
// msgpack value the reader can round-trip.
func exampleBody(v int) []byte {
b, err := msgpack.Marshal([]int{v})
if err != nil {
panic(err)
}
return b
}
func main() {
// A small source xlog to copy from.
var src bytes.Buffer
sw, err := writer.NewWriter(&src, exampleMeta())
if err != nil {
panic(err)
}
for lsn := int64(1); lsn <= 3; lsn++ {
if err := sw.WriteTx([]format.XRow{
{Type: iproto.IPROTO_INSERT, LSN: lsn, BodyRaw: exampleBody(int(lsn))},
}); err != nil {
panic(err)
}
}
if err := sw.Close(); err != nil {
panic(err)
}
// Read every record, cloning it (BodyRaw aliases the reader's buffer).
sr, err := reader.NewReader(bytes.NewReader(src.Bytes()))
if err != nil {
panic(err)
}
var block []format.XRow
for row, err := range sr.Rows() {
if err != nil {
panic(err)
}
clone := row
clone.BodyRaw = append([]byte(nil), row.BodyRaw...)
block = append(block, clone)
}
_ = sr.Close()
// Write all of them into one block of a new xlog.
var dst bytes.Buffer
dw, err := writer.NewWriter(&dst, exampleMeta())
if err != nil {
panic(err)
}
if err := dw.WriteBlock(block); err != nil {
panic(err)
}
if err := dw.Close(); err != nil {
panic(err)
}
fmt.Printf("copied %d records into one block\n", len(block))
}
Output: copied 3 records into one block
func (*Writer) WriteRawBlock ¶
WriteRawBlock writes a pre-framed tx block (fixheader + payload) verbatim: no row decode, no re-encode, no recompression, and no CRC recomputation. The block must be the exact on-disk bytes of a valid tx block — already carrying a correct fixheader and CRC32C — as produced by reader.NextBlockRaw. It is the write half of the verbatim block-copy fast path: a copy/truncate that does not transform rows forwards source blocks byte-for-byte, so a compressed (ZRow) block stays compressed without a decompress/recompress round trip.
Guards: ErrClosed after Close, ErrPendingWriteRow if a WriteRow accumulator is non-empty, and ErrInvalidRawBlock if block is shorter than a fixheader or is not prefixed by RowMarker / ZRowMarker. The block's CRC is trusted, not re-verified — it was validated when the block was read; corruption introduced between read and write is the caller's responsibility.
Example ¶
ExampleWriter_WriteRawBlock forwards physical tx blocks verbatim: each block's on-disk bytes are read with reader.NextBlockRaw and written unchanged with WriteRawBlock — no row decode, no re-encode, no recompression, no CRC recompute. This is the manual form of the fast copy path that pipe.CopyRaw wraps; reach for it when a tool needs to interleave verbatim block forwarding with other logic. A compressed block stays compressed on disk.
package main
import (
"bytes"
"errors"
"fmt"
"io"
"github.com/google/uuid"
"github.com/tarantool/go-iproto"
"github.com/vmihailenco/msgpack/v5"
"github.com/tarantool/go-xlog/format"
"github.com/tarantool/go-xlog/reader"
"github.com/tarantool/go-xlog/writer"
)
func exampleMeta() *format.Meta {
return &format.Meta{
Filetype: format.FiletypeXLOG,
Version: "example",
InstanceUUID: uuid.MustParse("11111111-2222-3333-4444-555555555555"),
}
}
// exampleBody msgpack-encodes a one-element tuple so the row body is a valid
// msgpack value the reader can round-trip.
func exampleBody(v int) []byte {
b, err := msgpack.Marshal([]int{v})
if err != nil {
panic(err)
}
return b
}
func main() {
// A small source xlog to copy from.
var srcBuf bytes.Buffer
sw, err := writer.NewWriter(&srcBuf, exampleMeta())
if err != nil {
panic(err)
}
for lsn := int64(1); lsn <= 3; lsn++ {
if err := sw.WriteTx([]format.XRow{
{Type: iproto.IPROTO_INSERT, LSN: lsn, BodyRaw: exampleBody(int(lsn))},
}); err != nil {
panic(err)
}
}
if err := sw.Close(); err != nil {
panic(err)
}
rd, err := reader.NewReader(bytes.NewReader(srcBuf.Bytes()))
if err != nil {
panic(err)
}
defer func() { _ = rd.Close() }()
var dstBuf bytes.Buffer
dw, err := writer.NewWriter(&dstBuf, exampleMeta())
if err != nil {
panic(err)
}
blocks := 0
for {
block, err := rd.NextBlockRaw()
if err != nil {
if errors.Is(err, io.EOF) {
break
}
panic(err)
}
// Block is valid only until the next NextBlockRaw call; WriteRawBlock
// consumes it immediately, so no copy is needed.
if err := dw.WriteRawBlock(block); err != nil {
panic(err)
}
blocks++
}
if err := dw.Close(); err != nil {
panic(err)
}
fmt.Printf("forwarded %d blocks verbatim\n", blocks)
}
Output: forwarded 3 blocks verbatim
func (*Writer) WriteRow ¶
WriteRow appends r to the writer's in-memory tx accumulator. The accumulator is flushed as a single tx block when either:
- the caller calls CommitTx, or
- r.IsCommit() is true at append time AND the buffered payload estimate is at or above AutocommitThreshold (a soft hint — a logical tx must not be split mid-stream, so we never auto-flush on a non-commit row).
To compose multi-row txs through this path, leave IPROTO_FLAG_COMMIT cleared on all but the last row, and either let WriteRow auto-flush on the last (if the caller pre-set IPROTO_FLAG_COMMIT) or call CommitTx explicitly. AssignTxIDs is applied during CommitTx — the caller does not need to manage TSN.
func (*Writer) WriteTx ¶
WriteTx is the all-in-one entry point: applies assignTxIDs, encodes the rows as one tx block, and writes them. It must not be mixed with a pending WriteRow accumulator — if the WriteRow buffer is non-empty, WriteTx returns an error rather than silently flushing (no implicit cross-API commits). Callers who want both APIs at once should explicitly CommitTx first.
Source Files