textmachine/platform/internal/ingest/tail_test.go

package ingest

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
)

// tailSink records what the tailer decided to materialize, and nothing else: what these tests are
// about is the READER's rule for at-least-once delivery, not what a sink does with an event.
type tailSink struct {
	hellos  []Hello
	applied []Envelope
	cursors []Cursor
	// failFrom makes the sink refuse every event from this seq on, which is how a database outage
	// mid-stream looks to the reader.
	failFrom int64
}

func (s *tailSink) Begin(_ context.Context, h Hello) error {
	s.hellos = append(s.hellos, h)
	return nil
}

func (s *tailSink) Apply(_ context.Context, ev Envelope, c Cursor) error {
	if s.failFrom > 0 && ev.Seq >= s.failFrom {
		return errors.New("database is down")
	}
	s.applied = append(s.applied, ev)
	s.cursors = append(s.cursors, c)
	return nil
}

func (s *tailSink) seqs() []int64 {
	out := make([]int64, 0, len(s.applied))
	for _, e := range s.applied {
		out = append(out, e.Seq)
	}
	return out
}

func line(t *testing.T, seq int64, typ Type, data any) string {
	t.Helper()
	raw, err := json.Marshal(data)
	if err != nil {
		t.Fatal(err)
	}
	b, err := json.Marshal(Envelope{Seq: seq, Type: typ, Data: raw})
	if err != nil {
		t.Fatal(err)
	}
	return string(b) + "\n"
}

func hello(t *testing.T, seq int64, runID string) string {
	return line(t, seq, TypeHello, Hello{StreamVersion: StreamVersion, EngineRunID: runID, BookID: "bk"})
}

func progress(t *testing.T, seq int64, done int) string {
	return line(t, seq, TypeProgress, Progress{Draft: Counter{Done: done, Total: 10}})
}

func journal(t *testing.T, content string) string {
	t.Helper()
	path := filepath.Join(t.TempDir(), JournalFile)
	if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
		t.Fatal(err)
	}
	return path
}

// The tailer is started when the unit is; the engine writes its first line whenever it gets there.
// Waiting is the correct behaviour and it must not be an error, or every run would log a failure
// during its first seconds — and today, with no emitter at all, forever.
func TestAnAbsentJournalIsNotAFailure(t *testing.T) {
	_, _, err := Tail(t.Context(), filepath.Join(t.TempDir(), JournalFile), "", Position{}, &tailSink{})
	if !errors.Is(err, ErrNoJournal) {
		t.Fatalf("missing journal gave %v, want ErrNoJournal", err)
	}
}

// A line without its newline is a line still being written. Reading it as an event turns an ordinary
// interleaving with the writer into a decode error on a paid run.
func TestAHalfWrittenLineIsLeftForNextTime(t *testing.T) {
	partial := hello(t, 1, "eng-1") + progress(t, 2, 1) + `{"seq":3,"type":"progress","dat`
	path := journal(t, partial)
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "", Position{}, sink)
	if err != nil {
		t.Fatal(err)
	}
	if id != "eng-1" {
		t.Fatalf("engine run id %q", id)
	}
	// The handshake moves the cursor like any other line: its effect is nothing, its seq is the
	// cursor, and without it a resume after the hello reads the next line as a gap.
	if got := sink.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Fatalf("applied %v, want 1,2", got)
	}
	// The cursor stands before the partial line, so the completed form of it is read next time.
	if want := int64(len(hello(t, 1, "eng-1")) + len(progress(t, 2, 1))); pos.Offset != want {
		t.Fatalf("offset %d, want %d (before the partial line)", pos.Offset, want)
	}
	// Now the writer finishes the line and appends another.
	rest := `a":{"draft":{"done":2,"total":10}}}` + "\n" + progress(t, 4, 3)
	f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o600)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := f.WriteString(rest); err != nil {
		t.Fatal(err)
	}
	f.Close()
	if _, _, err := Tail(t.Context(), path, id, pos, sink); err != nil {
		t.Fatal(err)
	}
	if got := sink.seqs(); len(got) != 4 || got[2] != 3 || got[3] != 4 {
		t.Fatalf("applied %v, want 1,2,3,4", got)
	}
}

// PD-105, ratified by D39.119: delivery is at-least-once, so re-reading the line the cursor stands
// on is NORMAL. It must be idempotent and it must not be fatal — the previous decoder called it
// ErrStreamGap, and after the PD-12 fix that killed a paid run over one duplicated line.
func TestARedeliveredLineIsNormalAndChangesNothing(t *testing.T) {
	body := hello(t, 1, "eng-1") + progress(t, 2, 1) + progress(t, 3, 2)
	path := journal(t, body)
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "", Position{}, sink)
	if err != nil {
		t.Fatal(err)
	}
	// Re-read the WHOLE journal from the beginning with the cursor already at seq 3: every line is a
	// duplicate, and not one of them may be applied again or raise an error.
	again := &tailSink{}
	pos2, _, err := Tail(t.Context(), path, id, Position{LastSeq: pos.LastSeq, LastHash: pos.LastHash}, again)
	if err != nil {
		t.Fatalf("re-reading an already applied journal: %v", err)
	}
	if len(again.applied) != 0 || len(again.hellos) != 0 {
		t.Fatalf("a re-read applied %d events and %d handshakes; both must be zero", len(again.applied), len(again.hellos))
	}
	if pos2.Offset != pos.Offset {
		t.Errorf("the byte hint did not catch up on a re-read: %d, want %d", pos2.Offset, pos.Offset)
	}
}

// The one case a high-water mark cannot absorb: the same seq saying something different. Quietly
// skipping it materializes whichever copy was read first (research/25 §эмиттер).
func TestTheSameSeqWithADifferentPayloadIsRefused(t *testing.T) {
	first := journal(t, hello(t, 1, "eng-1")+progress(t, 2, 1))
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), first, "", Position{}, sink)
	if err != nil {
		t.Fatal(err)
	}
	// The same seq, a different payload — a rewritten journal, or two engine instances on one run.
	rewritten := journal(t, hello(t, 1, "eng-1")+progress(t, 2, 7))
	if _, _, err := Tail(t.Context(), rewritten, id, Position{LastSeq: pos.LastSeq, LastHash: pos.LastHash}, sink); !errors.Is(err, ErrPayloadConflict) {
		t.Fatalf("a changed payload gave %v, want ErrPayloadConflict", err)
	}
}

// A gap means lines were lost, and reading further cannot recover them: the ratified answer is the
// resync channel, so the reader has to say so rather than materialize a projection with a hole.
func TestALostLineIsReportedRatherThanSkipped(t *testing.T) {
	path := journal(t, hello(t, 1, "eng-1")+progress(t, 2, 1)+progress(t, 5, 4))
	sink := &tailSink{}
	if _, _, err := Tail(t.Context(), path, "", Position{}, sink); !errors.Is(err, ErrStreamGap) {
		t.Fatalf("a skipped seq gave %v, want ErrStreamGap", err)
	}
	if got := sink.seqs(); len(got) != 2 || got[1] != 2 {
		t.Fatalf("applied %v: everything before the gap must still land", got)
	}
}

// The journal is per BOOK and append-only, so a resumed run appends a SECOND handshake with its own
// run id and a seq that starts over. A reader that treated it as corruption would stop exactly when
// the run resumed; one that adopted it would mix two attempts into one attempt's counters.
func TestAnotherAttemptsStreamInTheSameJournalIsSkipped(t *testing.T) {
	body := hello(t, 1, "eng-1") + progress(t, 2, 1) +
		hello(t, 1, "eng-2") + progress(t, 2, 5) + progress(t, 3, 6)
	path := journal(t, body)

	// Attempt one owns the first stream and never sees the second.
	first := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "", Position{}, first)
	if err != nil {
		t.Fatal(err)
	}
	if id != "eng-1" {
		t.Fatalf("adopted %q, want eng-1", id)
	}
	if got := first.seqs(); len(got) != 2 || got[1] != 2 {
		t.Fatalf("attempt one applied %v, want 1,2 of eng-1 and nothing of eng-2", got)
	}
	if pos.Offset != int64(len(body)) {
		t.Errorf("the reader did not walk past the other attempt's lines: offset %d of %d", pos.Offset, len(body))
	}

	// Attempt two starts where its own lines begin — the journal's size when it was admitted.
	start := int64(len(hello(t, 1, "eng-1") + progress(t, 2, 1)))
	second := &tailSink{}
	if _, id, err = Tail(t.Context(), path, "", Position{Offset: start}, second); err != nil {
		t.Fatal(err)
	}
	if id != "eng-2" {
		t.Fatalf("attempt two adopted %q, want eng-2", id)
	}
	if got := second.seqs(); len(got) != 3 || got[1] != 2 || got[2] != 3 {
		t.Fatalf("attempt two applied %v, want 1,2,3 of eng-2", got)
	}
}

// A journal that already holds SOMEBODY ELSE's stream at this attempt's starting offset — an
// operator running tmctl by hand in the book's directory is the realistic source.
//
// ⚠ Reproduced before the fix, and it was three defects in one line: the reader adopted the foreign
// handshake, materialized its events onto this attempt (a `ceiling` among them, which pauses a run
// nothing had paused), and then refused the attempt's OWN handshake for carrying a seq that was not
// 1 — quarantining the projection of a paying run for good. The cure is that the platform NAMES the
// stream before the unit starts (runs.engineStreamID) and hands the name over, so `want` is never
// empty and there is nothing left to adopt.
//
// Mutation caught: going back to `mine := want != "" && pos.Offset > 0`, and dropping the id from
// the spawn record so the reader is handed an empty one.
func TestAForeignStreamAtOurOffsetIsNeverAdopted(t *testing.T) {
	foreign := hello(t, 1, "SOMEONE-ELSE") + line(t, 2, TypeCeiling, Ceiling{Halted: true, Scope: ScopeBook})
	body := foreign + hello(t, 1, "ours") + progress(t, 2, 7)
	path := journal(t, body)

	// The attempt was admitted when the journal was empty, so its byte hint is 0 and it reads the
	// foreign lines on the way to its own.
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "ours", Position{}, sink)
	if err != nil {
		t.Fatalf("a foreign stream in the journal broke the read: %v", err)
	}
	if id != "ours" {
		t.Fatalf("the reader ended up on stream %q", id)
	}
	if len(sink.hellos) != 0 {
		t.Errorf("Begin was called for a stream this attempt did not own: %+v", sink.hellos)
	}
	if got := sink.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Fatalf("applied %v, want only our own 1,2 — a foreign ceiling event pauses a live run", got)
	}
	if pos.Offset != int64(len(body)) || pos.LastSeq != 2 {
		t.Errorf("cursor %+v after the whole journal", pos)
	}

	// And the case the byte HINT makes dangerous. An attempt's hint is the journal's SIZE when it was
	// admitted, so it is non-zero before the attempt has read a line — and if a foreign process is
	// writing at that moment, the hint lands INSIDE that process's stream, past its handshake. There
	// is then no hello above those lines to say whose they are.
	//
	// ⚠ This is why the reader asks about the CURSOR (`LastSeq > 0`, and the handshake is seq 1) and
	// not about the hint. Under "my offset is non-zero ⇒ these lines are mine" the foreign
	// continuation below is JUDGED against our cursor, its seq reads as a gap, and the projection of a
	// perfectly healthy attempt is quarantined for good.
	//
	// Mutation caught: `mine := want != "" && pos.Offset > 0`.
	before := hello(t, 1, "SOMEONE-ELSE") + progress(t, 2, 1)
	body = before + progress(t, 3, 2) + hello(t, 1, "ours") + progress(t, 2, 9)
	path = journal(t, body)
	restarted := &tailSink{}
	pos, _, err = Tail(t.Context(), path, "ours", Position{Offset: int64(len(before))}, restarted)
	if err != nil {
		t.Fatalf("a hint landing inside somebody else's stream broke the read: %v", err)
	}
	if got := restarted.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Errorf("applied %v, want only our own 1,2", got)
	}
	if pos.Offset != int64(len(body)) {
		t.Errorf("the reader did not walk past the foreign lines: offset %d of %d", pos.Offset, len(body))
	}
}

// The same journal read from the START by an attempt that already knows its own id. This is what a
// lost byte hint produces, and it is where the reader used to quarantine a perfectly healthy run:
// the previous attempt's seq 2 landed on our cursor's seq 2 with a different payload, which is
// exactly the corruption signal. Nothing but the last handshake says whose a line is.
func TestARereadFromTheStartDoesNotMistakeAnotherAttemptForCorruption(t *testing.T) {
	body := hello(t, 1, "eng-1") + progress(t, 2, 1) +
		hello(t, 1, "eng-2") + progress(t, 2, 5) + progress(t, 3, 6)
	path := journal(t, body)
	// Attempt two, mid-run: it knows its id and its cursor, and its byte hint is gone.
	own := &tailSink{}
	first, _, err := Tail(t.Context(), path, "eng-2", Position{}, own) // no byte hint at all
	if err != nil {
		t.Fatalf("re-reading from the start: %v", err)
	}
	if got := own.seqs(); len(got) != 3 || got[1] != 2 || got[2] != 3 {
		t.Fatalf("applied %v, want 1,2,3 of eng-2 and nothing of eng-1", got)
	}
	if first.Offset != int64(len(body)) {
		t.Errorf("offset %d, want %d", first.Offset, len(body))
	}
}

// Nothing is materialized before a handshake we own, even when the reader already knows its own
// engine run id. A journal that opens with events instead of a hello is the documented rot path
// (ad-hoc unversioned JSON), and judging those lines against OUR cursor would either apply another
// process's counters or report our own healthy stream as corrupt.
func TestEventsBeforeAnyHandshakeAreNotJudgedAgainstOurCursor(t *testing.T) {
	headless := progress(t, 2, 5) + progress(t, 3, 6) + hello(t, 1, "eng-1") + progress(t, 2, 1)
	path := journal(t, headless)
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "eng-1", Position{}, sink)
	if err != nil {
		t.Fatalf("a journal that opens without a handshake: %v", err)
	}
	if id != "eng-1" {
		t.Fatalf("engine run id %q", id)
	}
	if got := sink.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Fatalf("applied %v, want only the two lines that follow our own handshake", got)
	}
	if pos.LastSeq != 2 {
		t.Errorf("cursor at seq %d", pos.LastSeq)
	}
}

// The offset is a HINT and the append-only guarantee is what makes it usable. A journal that got
// SHORTER broke that guarantee, and reading on at a stale offset would decode the middle of a line.
func TestAJournalThatShrankIsRefusedRatherThanMisread(t *testing.T) {
	path := journal(t, hello(t, 1, "eng-1"))
	_, _, err := Tail(t.Context(), path, "eng-1", Position{Offset: 100_000, LastSeq: 5}, &tailSink{})
	if err == nil || !strings.Contains(err.Error(), "append-only") {
		t.Fatalf("a shrunken journal gave %v, want a refusal naming the broken guarantee", err)
	}
}

// The cursor the sink is handed has to be the position AFTER the line, or a crash between the effect
// and the next read re-reads a line the sink has already applied — harmless by luck, not by design.
func TestTheCursorHandedToTheSinkIsPastTheLine(t *testing.T) {
	h, p2 := hello(t, 1, "eng-1"), progress(t, 2, 1)
	path := journal(t, h+p2)
	sink := &tailSink{}
	if _, _, err := Tail(t.Context(), path, "", Position{}, sink); err != nil {
		t.Fatal(err)
	}
	if len(sink.cursors) != 2 {
		t.Fatalf("%d cursors, want one per applied line", len(sink.cursors))
	}
	if want := int64(len(h)); sink.cursors[0].Offset != want {
		t.Errorf("handshake cursor offset %d, want %d", sink.cursors[0].Offset, want)
	}
	if want := int64(len(h) + len(p2)); sink.cursors[1].Offset != want {
		t.Errorf("cursor offset %d, want %d", sink.cursors[1].Offset, want)
	}
	if len(sink.cursors[1].SHA256) != 32 {
		t.Errorf("cursor carries no line hash: %v", sink.cursors[1].SHA256)
	}
}

// A stream whose major version changed means something different by the same words; materializing it
// is worse than not materializing it.
func TestAMajorVersionBumpIsRefused(t *testing.T) {
	body, err := json.Marshal(Envelope{Seq: 1, Type: TypeHello,
		Data: json.RawMessage(`{"stream_version":"9.9","engine_run_id":"eng-1"}`)})
	if err != nil {
		t.Fatal(err)
	}
	path := journal(t, string(body)+"\n")
	if _, _, err := Tail(t.Context(), path, "", Position{}, &tailSink{}); !errors.Is(err, ErrUnsupportedVersion) {
		t.Fatalf("a major bump gave %v, want ErrUnsupportedVersion", err)
	}
}

// Half of the ratified idempotency key lives in the handshake; without it every run's events would
// collapse into one namespace instead of failing.
func TestAHandshakeWithoutAnEngineRunIdIsRefused(t *testing.T) {
	body := fmt.Sprintf(`{"seq":1,"type":"hello","data":{"stream_version":%q,"engine_run_id":""}}`+"\n", StreamVersion)
	path := journal(t, body)
	if _, _, err := Tail(t.Context(), path, "", Position{}, &tailSink{}); !errors.Is(err, ErrBadHandshake) {
		t.Fatalf("an anonymous handshake gave %v, want ErrBadHandshake", err)
	}
}

// A sink that fails stops the read where it is: the cursor must not advance past an effect that was
// never applied.
func TestAFailingSinkDoesNotMoveTheCursor(t *testing.T) {
	path := journal(t, hello(t, 1, "eng-1")+progress(t, 2, 1)+progress(t, 3, 2))
	sink := &tailSink{failFrom: 2}
	pos, _, err := Tail(t.Context(), path, "", Position{}, sink)
	if err == nil {
		t.Fatal("a failing sink was not reported")
	}
	if pos.LastSeq != 1 {
		t.Errorf("cursor moved to seq %d past an event the sink refused", pos.LastSeq)
	}
}

// A line past the buffer is REFUSED, not swallowed. Returning io.EOF for it would make the tailer
// report "caught up" at that byte forever: nothing after it would ever be materialized, with no
// error and no quarantine.
func TestALinePastTheBufferIsRefusedRatherThanRead(t *testing.T) {
	huge := `{"seq":2,"type":"progress","data":{"note":"` + strings.Repeat("x", maxLine) + `"}}`
	path := journal(t, hello(t, 1, "eng-1")+huge+"\n")
	_, _, err := Tail(t.Context(), path, "", Position{}, &tailSink{})
	if err == nil || !strings.Contains(err.Error(), "exceeds") {
		t.Fatalf("an over-long line gave %v, want a refusal naming the limit", err)
	}
}

// A blank line is not an event, but it IS bytes: an offset that did not advance past it is short by
// its length from then on, permanently.
func TestABlankLineStillMovesTheOffset(t *testing.T) {
	body := hello(t, 1, "eng-1") + "\n" + progress(t, 2, 1)
	path := journal(t, body)
	pos, _, err := Tail(t.Context(), path, "", Position{}, &tailSink{})
	if err != nil {
		t.Fatal(err)
	}
	if pos.Offset != int64(len(body)) {
		t.Errorf("offset %d after a journal of %d bytes containing a blank line", pos.Offset, len(body))
	}
}

// Neither half of the cursor may advance past an effect the sink refused — the byte hint as much as
// the seq, because the hint is what decides where the next sweep starts reading.
func TestNeitherHalfOfTheCursorAdvancesPastARefusedEffect(t *testing.T) {
	h, p2 := hello(t, 1, "eng-1"), progress(t, 2, 1)
	path := journal(t, h+p2+progress(t, 3, 2))
	sink := &tailSink{failFrom: 3}
	pos, _, err := Tail(t.Context(), path, "", Position{}, sink)
	if err == nil {
		t.Fatal("a failing sink was not reported")
	}
	if pos.LastSeq != 2 {
		t.Errorf("cursor seq %d past a refused event", pos.LastSeq)
	}
	if want := int64(len(h) + len(p2)); pos.Offset != want {
		t.Errorf("cursor offset %d, want %d — the byte hint moved past a refused event", pos.Offset, want)
	}
}