ingot/internal/head/appender.go

package head

import (
	"errors"

	"git.dvdt.dev/david/ingot/internal/wal"
	"git.dvdt.dev/david/ingot/labels"
)

var (
	ErrOutOfOrder     = errors.New("head: out-of-order sample")
	ErrSeriesNotFound = errors.New("head: unknown series ref")
	ErrAppenderClosed = errors.New("head: appender already closed")
)

// Appender buffers samples and new series, then atomically commits them
// to the WAL and applies them to the head.
type Appender struct {
	head *Head

	// New series created during this batch (not yet in WAL).
	newSeries []wal.SeriesRecord
	newHashes []uint64 // parallel to newSeries, for rollback

	// Buffered samples.
	samples []wal.RefSample

	// Last timestamp seen per series in this batch, for OOO rejection
	// of samples that haven't been committed yet.
	batchLastT map[uint64]int64

	closed bool
}

// Append adds a sample to the batch. If ref is 0, the series is resolved
// (or created) from ls. Returns the series ref for fast-path reuse.
func (a *Appender) Append(ref uint64, ls []labels.Label, t int64, v float64) (uint64, error) {
	if ref == 0 {
		return a.appendByLabels(ls, t, v)
	}
	return a.appendByRef(ref, t, v)
}

func (a *Appender) appendByLabels(ls []labels.Label, t int64, v float64) (uint64, error) {
	ls = labels.Sort(ls)
	if err := labels.Validate(ls); err != nil {
		return 0, err
	}

	hash := labels.Hash(ls)
	s := a.head.series.getByHash(hash, ls)

	if s == nil {
		// New series.
		ref := a.head.nextRef.Add(1)
		s = &memSeries{
			ref:    ref,
			labels: copyLabels(ls),
		}
		a.head.series.set(hash, s)
		a.newSeries = append(a.newSeries, wal.SeriesRecord{Ref: ref, Labels: s.labels})
		a.newHashes = append(a.newHashes, hash)
	}

	if err := a.checkTimestamp(s.ref, t); err != nil {
		return 0, err
	}

	a.samples = append(a.samples, wal.RefSample{Ref: s.ref, T: t, V: v})
	return s.ref, nil
}

func (a *Appender) appendByRef(ref uint64, t int64, v float64) (uint64, error) {
	s := a.head.series.getByRef(ref)
	if s == nil {
		return 0, ErrSeriesNotFound
	}

	if err := a.checkTimestamp(ref, t); err != nil {
		return 0, err
	}

	a.samples = append(a.samples, wal.RefSample{Ref: ref, T: t, V: v})
	return ref, nil
}

// checkTimestamp validates that t is strictly after the last sample for
// this series, both in committed state and in the current batch.
func (a *Appender) checkTimestamp(ref uint64, t int64) error {
	// Check committed state.
	s := a.head.series.getByRef(ref)
	if s != nil {
		s.mu.Lock()
		lastT, hasData := s.lastT, s.hasData
		s.mu.Unlock()
		if hasData && t <= lastT {
			return ErrOutOfOrder
		}
	}
	// Check batch state.
	if a.batchLastT == nil {
		a.batchLastT = make(map[uint64]int64)
	}
	if prev, ok := a.batchLastT[ref]; ok && t <= prev {
		return ErrOutOfOrder
	}
	a.batchLastT[ref] = t
	return nil
}

// Commit writes the batch to the WAL, then applies it to the head.
func (a *Appender) Commit() error {
	if a.closed {
		return ErrAppenderClosed
	}
	a.closed = true

	// WAL: series records first, then samples.
	if len(a.newSeries) > 0 {
		if err := a.head.wal.LogSeries(a.newSeries); err != nil {
			return err
		}
	}
	if len(a.samples) > 0 {
		if err := a.head.wal.LogSamples(a.samples); err != nil {
			return err
		}
	}

	// Apply samples to head.
	for _, s := range a.samples {
		a.head.applySample(s.Ref, s.T, s.V)
	}

	return nil
}

// Rollback discards the batch. Any new series created during this batch
// that have no data are removed.
func (a *Appender) Rollback() error {
	if a.closed {
		return ErrAppenderClosed
	}
	a.closed = true

	// Remove new series that were registered but never committed.
	for i, rec := range a.newSeries {
		s := a.head.series.getByRef(rec.Ref)
		if s != nil && !s.hasData {
			a.head.series.remove(a.newHashes[i], s)
		}
	}

	a.newSeries = nil
	a.newHashes = nil
	a.samples = nil
	return nil
}

func copyLabels(ls []labels.Label) []labels.Label {
	c := make([]labels.Label, len(ls))
	copy(c, ls)
	return c
}