Documentation
¶
Index ¶
- Constants
- Variables
- func CleanSourcePath(src *slog.Source)
- func CreateFile(filename string, fi os.FileInfo) (*os.File, error)
- func DecryptIfSealed(rd io.Reader, decrypt func(io.Reader) (io.Reader, error)) (io.Reader, error)
- func ExtractLTXTimestamp(rd io.Reader) (time.Time, io.Reader, error)
- func Fileinfo(fi os.FileInfo) (uid, gid int)
- func Hexdump(data []byte) string
- func InitLog(w io.Writer, level, typ string, addSource bool)
- func LimitReadCloser(r io.ReadCloser, n int64) io.ReadCloser
- func LockFileExclusive(f *os.File) error
- func MkdirAll(path string, fi os.FileInfo) error
- func NewLZ4Reader(r io.ReadCloser) io.ReadCloser
- func ReplaceAttr(groups []string, a slog.Attr) slog.Attr
- func UnlockFile(f *os.File) error
- type LTXFileOpener
- type LZ4ReadCloser
- type LimitedReadCloser
- type LoggingConfig
- type ReadCloser
- type ReadCounter
- type ResumableReader
Constants ¶
const LevelTrace = slog.LevelDebug - 4
Variables ¶
var ( OperationTotalCounterVec = metric.NewCounterVec(metric.CounterOpts{ Name: "replicate_replica_operation_total", Help: "The number of replica operations performed", }, []string{"replica_type", "operation"}) OperationBytesCounterVec = metric.NewCounterVec(metric.CounterOpts{ Name: "replicate_replica_operation_bytes", Help: "The number of bytes used by replica operations", }, []string{"replica_type", "operation"}) OperationDurationHistogramVec = metric.NewHistogramVec(metric.HistogramOpts{ Name: "replicate_replica_operation_duration_seconds", Help: "Duration of replica operations by type and operation", Buckets: []float64{0.01, 0.05, 0.1, 0.5, 1, 5, 10, 30, 60}, }, []string{"replica_type", "operation"}) OperationErrorCounterVec = metric.NewCounterVec(metric.CounterOpts{ Name: "replicate_replica_operation_errors_total", Help: "Number of replica operation errors by type, operation, and error code", }, []string{"replica_type", "operation", "code"}) L0RetentionGaugeVec = metric.NewGaugeVec(metric.GaugeOpts{ Name: "replicate_l0_retention_files_total", Help: "Number of L0 files by status during retention enforcement", }, []string{"db", "status"}) )
Shared replica metrics.
Functions ¶
func CleanSourcePath ¶
func CreateFile ¶
CreateFile creates the file and matches the mode & uid/gid of fi.
func DecryptIfSealed ¶ added in v0.9.9
DecryptIfSealed returns a reader of plaintext LTX bytes from rd, decrypting only when rd actually IS an age ciphertext.
Why this is not simply age.Decrypt: having identities configured is not the same fact as the object being sealed. A bucket can hold both — ours does, because snapshots were written before age was wired and sealed ones landed after — and decrypting unconditionally turns every older object into:
age decrypt: failed to read header: parsing age header: unexpected intro "LTX1\x00…"
That took chat, hanzo-app, dataroom and studio down: the restore init container cannot rehydrate an empty PVC, so no pod with this sidecar can cold-start. There was no config that read both, because removing the identities makes the WRITER fail closed instead.
The write path already made exactly this distinction — ExtractLTXTimestamp above sniffs the same intro so that mixed buckets upload correctly. This is the read half of that, so the two directions agree.
It does NOT weaken encryption: with zero identities the caller never had a key and this returns the stream untouched, exactly as before. A sealed object still requires a valid identity, and a corrupt one still fails — only the plaintext case, which previously could not be read at all, now round-trips.
func ExtractLTXTimestamp ¶ added in v0.9.1
ExtractLTXTimestamp returns the timestamp to record for an object being written to a replica, along with a reader that replays the full stream (the header bytes consumed for inspection are preserved, so callers upload the complete object).
For a plaintext LTX object the timestamp is read from the LTX header. For an age-encrypted object the header is unreadable, so the current time is used — this is what makes client-side age encryption work with every storage backend, all of which otherwise fail trying to parse an LTX header out of ciphertext ("extract timestamp from LTX header: invalid LTX file"). A stream that is neither a valid LTX nor age ciphertext is a genuine corrupt-object error and is surfaced as before.
func LimitReadCloser ¶
func LimitReadCloser(r io.ReadCloser, n int64) io.ReadCloser
Copied from the io package to implement io.Closer.
func LockFileExclusive ¶
LockFileExclusive forwards to dblock.LockExclusive so the replica applier and external apply callers (shard layer) share one canonical SQLite-byte-range lock implementation.
func MkdirAll ¶
MkdirAll is a copy of os.MkdirAll() except that it attempts to set the mode/uid/gid to match fi for each created directory.
func NewLZ4Reader ¶
func NewLZ4Reader(r io.ReadCloser) io.ReadCloser
NewLZ4Reader creates an LZ4 decompressing reader that wraps the source. Closing the returned reader also closes the underlying source.
Types ¶
type LTXFileOpener ¶
type LZ4ReadCloser ¶
LZ4ReadCloser wraps an LZ4 reader with the underlying source for proper closing.
func (*LZ4ReadCloser) Close ¶
func (r *LZ4ReadCloser) Close() error
type LimitedReadCloser ¶
type LimitedReadCloser struct {
R io.ReadCloser // underlying reader
N int64 // max bytes remaining
}
func (*LimitedReadCloser) Close ¶
func (l *LimitedReadCloser) Close() error
type LoggingConfig ¶
type LoggingConfig struct {
Level string `yaml:"level"`
Type string `yaml:"type"`
Stderr bool `yaml:"stderr"`
Source bool `yaml:"source"`
}
LoggingConfig configures logging.
type ReadCloser ¶
type ReadCloser struct {
// contains filtered or unexported fields
}
ReadCloser wraps a reader to also attach a separate closer.
func NewReadCloser ¶
func NewReadCloser(r io.Reader, c io.Closer) *ReadCloser
NewReadCloser returns a new instance of ReadCloser.
func (*ReadCloser) Close ¶
func (r *ReadCloser) Close() error
Close closes the reader (if implementing io.ReadCloser) and the Closer.
type ReadCounter ¶
type ReadCounter struct {
// contains filtered or unexported fields
}
ReadCounter wraps an io.Reader and counts the total number of bytes read.
func NewReadCounter ¶
func NewReadCounter(r io.Reader) *ReadCounter
NewReadCounter returns a new instance of ReadCounter that wraps r.
type ResumableReader ¶
type ResumableReader struct {
// contains filtered or unexported fields
}
resumableReader wraps an io.ReadCloser from a remote storage backend with automatic reconnection on read errors.
During restore, the LTX compactor opens all LTX file streams upfront, then processes pages in page-number order. Incremental LTX files that only contain high-numbered pages may have their S3/storage streams sit idle for minutes while the compactor works through lower-numbered pages from the snapshot. Storage providers (S3, Tigris, etc.) may close these idle connections, causing "unexpected EOF" errors.
This reader detects two failure modes:
- Non-EOF errors (connection reset, timeout) - the stream broke mid-transfer.
- Premature EOF - the server closed the connection cleanly, but we haven't read all bytes yet (detected by comparing offset against known file size).
On failure, it closes the dead stream and reopens from the current byte offset using the storage backend's range request support (the offset parameter of OpenLTXFile). Callers like io.ReadFull see a seamless byte stream because partial reads are returned without error, prompting the caller to request remaining bytes on the next Read call.
func NewResumableReader ¶
func NewResumableReader(ctx context.Context, client LTXFileOpener, level int, minTXID, maxTXID ltx.TXID, size int64, rc io.ReadCloser, logger *slog.Logger) *ResumableReader
NewResumableReader creates a ResumableReader. Primarily exposed for testing.
func (*ResumableReader) Close ¶
func (r *ResumableReader) Close() error
Source Files
Directories