textmachine/platform/internal/ingest/tail.go

package ingest

import (
	"bufio"
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
)

// JournalFile is the engine's event journal, in the BOOK's directory (D39.106 §2).
const JournalFile = "events.jsonl"

var (
	// ErrPayloadConflict is the same seq carrying a different payload. Re-reading a line is normal;
	// re-reading it and finding it CHANGED is not, and it cannot be reconciled by reading on —
	// either the journal was rewritten or two engine instances overlapped on one run id
	// (research/25 §эмиттер). The attempt is quarantined rather than materialized on a guess.
	ErrPayloadConflict = errors.New("ingest: a seq was re-read with a different payload")
	// ErrNoJournal is a journal that does not exist yet. Not a failure: the tailer is started when
	// the unit is, and the engine writes its first line whenever it gets there.
	ErrNoJournal = errors.New("ingest: journal not present yet")
)

// Position is where the reader stands in one journal.
//
// LastSeq is half of the RATIFIED cursor (engine_run_id, seq); Offset is a HINT, and the difference
// is not pedantry — a hint may be wrong and is then discarded, whereas the cursor decides whether an
// event has already been applied. Both are committed in the SAME transaction as the effect, which
// is what makes at-least-once delivery safe to re-read.
type Position struct {
	Offset   int64
	LastSeq  int64
	LastHash []byte
}

// Cursor travels with an event so a sink can persist the two together.
type Cursor struct {
	// Offset is the byte position just past the line this event came from.
	Offset int64
	// SHA256 of the raw line. Stored for the highest applied seq only: it is what turns "I have seen
	// this seq" into "I have seen this seq and it said the same thing" (PD-105).
	SHA256 []byte
}

// Tail applies every COMPLETE line the journal has gained since `from`, and returns where it now
// stands. It returns ErrNoJournal while the file is absent, which the caller treats as "not yet".
//
// A partial trailing line is deliberately left alone: the writer appends, and a line without its
// newline is a line still being written, not a malformed one. Reading it would turn an ordinary
// interleaving into a decode error on a paid run.
//
// want is the engine run id this attempt owns, which the platform CHOOSES before the unit is created
// (runs.engineStreamID). Lines belonging to any other engine run id are skipped: the journal is per
// BOOK and append-only, so a resumed run appends a second hello to the same file, and a reader that
// treated it as a corruption would stop exactly when the run resumed.
//
// ⚠ Empty means the attempt was started before the platform named its stream, and then — and only
// then — the first hello at or after the starting offset is ADOPTED. That fallback is the shape of a
// measured defect and is kept only for attempts that predate the naming: a journal is per book, so
// "the first hello at my offset" is whatever was written there, and a foreign one had its events
// materialized onto this attempt and then quarantined the projection when the real handshake
// arrived at a seq that was not 1.
func Tail(ctx context.Context, path, want string, from Position, sink Sink) (Position, string, error) {
	f, err := os.Open(path)
	if errors.Is(err, fs.ErrNotExist) {
		return from, want, ErrNoJournal
	}
	if err != nil {
		return from, want, fmt.Errorf("ingest: open journal: %w", err)
	}
	defer f.Close()

	// A journal shorter than our hint is a journal that was replaced, which the ratified form says
	// cannot happen (append-only). Rather than read garbage at a stale offset, start over and say so.
	if st, err := f.Stat(); err == nil && st.Size() < from.Offset {
		return Position{}, want, fmt.Errorf("ingest: journal %s shrank from %d to %d bytes: it is not append-only", path, from.Offset, st.Size())
	}
	if from.Offset > 0 {
		if _, err := f.Seek(from.Offset, io.SeekStart); err != nil {
			return from, want, fmt.Errorf("ingest: seek journal: %w", err)
		}
	}
	return tailFrom(ctx, f, want, from, sink)
}

func tailFrom(ctx context.Context, r io.Reader, want string, pos Position, sink Sink) (Position, string, error) {
	br := bufio.NewReaderSize(r, maxLine)
	// Which stream the current lines belong to. A line carries a seq and NO stream id, so the only
	// thing that says whose it is, is the last handshake above it. We are inside our own region when
	// we have already APPLIED lines of it — which is what a non-zero last_seq means, the handshake
	// itself being seq 1.
	//
	// ⚠ It used to ask about the byte OFFSET instead, and that was wrong at exactly one moment: a new
	// attempt starts at the journal's current SIZE, so its offset is non-zero before it has read a
	// thing, and every line written there was assumed to be ours. Anything a foreign process appended
	// in that window — an operator's own tmctl in the book's directory — was then materialized onto
	// this attempt, `ceiling` events included.
	//
	// Getting this wrong is not a subtle loss in either direction: a previous attempt's seq 2 read
	// against our cursor's seq 2 is a different payload at the same number, i.e. exactly the
	// corruption signal, so a resumed run quarantined itself the first time the reader walked the
	// journal from the start.
	mine := want != "" && pos.LastSeq > 0
	for {
		if err := ctx.Err(); err != nil {
			return pos, want, err
		}
		line, err := readLine(br)
		if errors.Is(err, io.EOF) {
			return pos, want, nil // caught up; the rest of the line, if any, is still being written
		}
		if err != nil {
			return pos, want, err
		}
		// Where the cursor stands ONCE this line has been applied. It is what the sink persists, in
		// the same transaction as the effect, so a crash between the two cannot skip the line.
		at := pos.Offset + int64(len(line))
		body := bytes.TrimRight(line, "\n")
		if len(bytes.TrimSpace(body)) == 0 {
			pos.Offset = at // a blank line is not an event
			continue
		}
		var ev Envelope
		if err := json.Unmarshal(body, &ev); err != nil {
			return pos, want, fmt.Errorf("ingest: malformed line at offset %d: %w", pos.Offset, err)
		}
		next, id, own, err := apply(ctx, ev, body, want, mine,
			Position{Offset: at, LastSeq: pos.LastSeq, LastHash: pos.LastHash}, sink)
		if err != nil {
			return pos, want, err
		}
		pos, want, mine = next, id, own
	}
}

// apply decides what one decoded line means for the cursor and hands it to the sink. pos.Offset is
// already the position PAST this line, so whatever it returns is where the reader now stands. mine
// says whether the lines at this point belong to the attempt being tailed.
func apply(ctx context.Context, ev Envelope, body []byte, want string, mine bool, pos Position, sink Sink) (Position, string, bool, error) {
	sum := sha256.Sum256(body)
	held := Position{Offset: pos.Offset, LastSeq: pos.LastSeq, LastHash: pos.LastHash}
	if ev.Type == TypeHello {
		var h Hello
		if err := json.Unmarshal(ev.Data, &h); err != nil {
			return pos, want, mine, fmt.Errorf("ingest: hello payload: %w", err)
		}
		if err := checkVersion(h.StreamVersion); err != nil {
			return pos, want, mine, err
		}
		if h.EngineRunID == "" {
			return pos, want, mine, fmt.Errorf("%w: no engine_run_id", ErrBadHandshake)
		}
		// The handshake is seq 1 by definition, and here it also MOVES the cursor — so a hello at seq 0
		// would leave the cursor at zero and the first real event would then be dropped as a duplicate
		// of it. The pipe decoder has always required this (decoder.go); the file reader must too.
		if ev.Seq != 1 {
			return pos, want, mine, fmt.Errorf("%w: hello carries seq %d, want 1", ErrBadHandshake, ev.Seq)
		}
		switch {
		case want == "":
			// The handshake of the attempt this tailer was started for.
			if err := sink.Begin(ctx, h); err != nil {
				return pos, want, mine, err
			}
			want, mine = h.EngineRunID, true
		case h.EngineRunID == want:
			mine = true // our own handshake, read again
		default:
			return held, want, false, nil // another attempt's stream begins here
		}
		// and then it falls through to the ordinary rules below. The handshake is seq 1 of the stream,
		// so it MOVES THE CURSOR like any other line — and it has to, or the byte hint outlives a
		// last_seq that stayed at zero and the very next line reads as a gap. Its effect on the read
		// model is nothing; its effect on the cursor is the point.
	}
	if want == "" || !mine {
		// Either no handshake of ours has been seen yet, or these lines belong to another attempt of
		// the same book — the journal is per BOOK and append-only, so both are ordinary. Nothing is
		// materialized and, crucially, nothing is JUDGED: a foreign seq compared against our cursor
		// reads as a payload conflict at the same number.
		return held, want, mine, nil
	}
	switch {
	case ev.Seq == pos.LastSeq:
		// At-least-once delivery: re-reading the line the cursor stands on is NORMAL (PD-105, ratified
		// by D39.119) and must be idempotent. The same seq with a DIFFERENT payload is the one case a
		// high-water mark cannot absorb.
		if len(pos.LastHash) > 0 && !bytes.Equal(pos.LastHash, sum[:]) {
			return pos, want, mine, fmt.Errorf("%w: seq %d", ErrPayloadConflict, ev.Seq)
		}
		return held, want, mine, nil
	case ev.Seq < pos.LastSeq:
		// Already applied and already surpassed. Only the highest seq keeps a hash, so this one rests
		// on the journal's append-only guarantee rather than being verified — stated because an
		// unstated limit is the kind that gets read as a check.
		return held, want, mine, nil
	case ev.Seq > pos.LastSeq+1:
		return pos, want, mine, fmt.Errorf("%w: seq %d after %d", ErrStreamGap, ev.Seq, pos.LastSeq)
	}
	if err := sink.Apply(ctx, ev, Cursor{Offset: pos.Offset, SHA256: sum[:]}); err != nil {
		return pos, want, mine, err
	}
	return Position{Offset: pos.Offset, LastSeq: ev.Seq, LastHash: sum[:]}, want, mine, nil
}

// readLine returns one line INCLUDING its newline, or io.EOF when no complete line is available. A
// line longer than the buffer is refused rather than grown: events carry counters and ids, never
// text, so a megabyte line means the journal is not what we think it is.
//
// ⚠ Only a real end of file means "caught up". Mapping EVERY error to io.EOF — which the first
// version did — turns a failing disk into "nothing new": materialization would stop and no sweep
// would ever say why.
func readLine(br *bufio.Reader) ([]byte, error) {
	line, err := br.ReadSlice('\n')
	switch {
	case err == nil:
		return line, nil
	case errors.Is(err, bufio.ErrBufferFull):
		return nil, fmt.Errorf("ingest: journal line exceeds %d bytes", maxLine)
	case errors.Is(err, io.EOF):
		// Includes a trailing partial line: it is not finished being written.
		return nil, io.EOF
	default:
		return nil, fmt.Errorf("ingest: read journal: %w", err)
	}
}