ndjson

package
v0.1.37 Latest Latest
Warning

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

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

Documentation

Overview

Package ndjson stores each record stream as a local file: one line per row in <dir>/<kind>/<stream>.ndjson, written {"seq":N,"row":{…}}, beside a <stream>.meta.json sidecar holding the stream's recordstore.Meta. A trimmed stream's rows move to <stream>@<low seq>.ndjson, which the sidecar names.

It is the durable backend a CLI run writes when no shared store is configured. Two bounds keep it from filling a disk nobody watches: a per stream byte cap, which refuses an append loudly rather than dropping it, and a per kind count of streams kept, which removes the oldest when a new one opens.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Backend

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

Backend is a recordstore.Backend over a directory of files.

func New

func New(options Options) (*Backend, error)

New validates options and returns the backend.

func (*Backend) Append

func (b *Backend) Append(ctx context.Context, stream, kind string, rows []recordstore.Row) (recordstore.AppendResult, error)

Append encodes the rows it keeps as lines after the stream's high seq and appends them.

func (*Backend) Close

func (b *Backend) Close() error

Close releases nothing: every call opens and closes its own files.

func (*Backend) Expire

func (b *Backend) Expire(_ context.Context, stream string, ttl time.Duration) error

Expire moves stream's expiry to ttl from now.

func (*Backend) Meta

func (b *Backend) Meta(_ context.Context, stream string) (recordstore.Meta, error)

Meta describes stream.

func (*Backend) Scan

func (b *Backend) Scan(_ context.Context, stream string, afterSeq int64, fn func(int64, recordstore.Row) error) error

Scan reads stream's file in seq order up to its committed high seq, starting at the first line after afterSeq rather than at the top of the file.

func (*Backend) Trim

func (b *Backend) Trim(_ context.Context, stream string, before time.Time) (recordstore.Meta, error)

Trim removes the rows appended before before.

type Options

type Options struct {
	// Dir holds a directory per kind.
	Dir string

	// Schema resolves a kind to its key and retention. A kind it refuses
	// cannot be written.
	Schema recordstore.SchemaResolver

	// MaxBytes caps one stream's file. An append that would take it past the
	// cap is refused whole with recordstore.ErrCapacity and the stream is
	// marked capped.
	MaxBytes int64

	// KeepStreams is how many streams of one kind the directory keeps. Opening
	// a new stream removes the least recently written beyond it; a stream
	// being written at that moment is left alone.
	KeepStreams int

	// TTL is how long a stream lives from its first append unless Expire moves
	// it, or how long a row of a kind retaining rows lives from its own append.
	// Zero keeps a stream until it is expired or rotated out, and refuses a
	// kind retaining rows.
	TTL time.Duration

	// Now is the clock streams are stamped and expired by. Nil is time.Now.
	Now func() time.Time
}

Options configure an ndjson backend.

Jump to

Keyboard shortcuts

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