119 lines
4.1 KiB
Go
119 lines
4.1 KiB
Go
package ingest
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
)
|
|
|
|
var (
|
|
// ErrUnsupportedVersion is a major-version refusal: materializing a stream whose meaning
|
|
// changed is worse than not materializing it.
|
|
ErrUnsupportedVersion = errors.New("ingest: unsupported stream version")
|
|
// ErrNoHandshake is a stream whose first line is not the handshake. Ad-hoc unversioned JSON is
|
|
// the documented rot path (docker jsonmessage) and is refused at the door.
|
|
ErrNoHandshake = errors.New("ingest: stream does not open with hello")
|
|
// ErrStreamGap is a seq that skipped or went backwards: lines were lost. Recovery is not
|
|
// guesswork — the caller reconciles from `tmctl status --json`, the ratified resync channel.
|
|
ErrStreamGap = errors.New("ingest: sequence gap")
|
|
// ErrBadHandshake is a handshake that parses but does not identify the stream. Half of the
|
|
// ratified idempotency key lives in it, so an empty engine_run_id would collapse every run's
|
|
// events into one namespace rather than fail (PD-10a).
|
|
ErrBadHandshake = errors.New("ingest: incomplete handshake")
|
|
// ErrRepeatedHello is a second handshake mid-stream: the process on the other end restarted, or
|
|
// two streams were spliced. Either way the identity behind the seq numbers changed, and the
|
|
// version gate only ever inspected line 1 (PD-10c).
|
|
ErrRepeatedHello = errors.New("ingest: hello after the handshake")
|
|
)
|
|
|
|
// maxLine caps one event. Events carry counters and ids, never text — the translated text travels
|
|
// as artifacts — so a megabyte line means the stream is not what we think it is.
|
|
const maxLine = 1 << 20
|
|
|
|
// Decoder reads the NDJSON event stream.
|
|
type Decoder struct {
|
|
sc *bufio.Scanner
|
|
hello Hello
|
|
greeted bool
|
|
lastSeq int64
|
|
}
|
|
|
|
func NewDecoder(r io.Reader) *Decoder {
|
|
sc := bufio.NewScanner(r)
|
|
sc.Buffer(make([]byte, 0, 64<<10), maxLine)
|
|
return &Decoder{sc: sc}
|
|
}
|
|
|
|
// Hello reads and validates the handshake. It must be called before Next.
|
|
func (d *Decoder) Hello() (Hello, error) {
|
|
ev, err := d.line()
|
|
if err != nil {
|
|
if errors.Is(err, io.EOF) {
|
|
return Hello{}, ErrNoHandshake
|
|
}
|
|
return Hello{}, err
|
|
}
|
|
if ev.Type != TypeHello {
|
|
return Hello{}, fmt.Errorf("%w: first line is %q", ErrNoHandshake, ev.Type)
|
|
}
|
|
// The handshake is seq 1 by definition. Without this check a stream whose first lines were lost
|
|
// still opens, and the loss is undetectable: the gap check below only compares neighbours.
|
|
if ev.Seq != 1 {
|
|
return Hello{}, fmt.Errorf("%w: hello carries seq %d, want 1", ErrStreamGap, ev.Seq)
|
|
}
|
|
var h Hello
|
|
if err := json.Unmarshal(ev.Data, &h); err != nil {
|
|
return Hello{}, fmt.Errorf("ingest: hello payload: %w", err)
|
|
}
|
|
if err := checkVersion(h.StreamVersion); err != nil {
|
|
return Hello{}, err
|
|
}
|
|
if h.EngineRunID == "" {
|
|
return Hello{}, fmt.Errorf("%w: no engine_run_id", ErrBadHandshake)
|
|
}
|
|
d.hello, d.greeted = h, true
|
|
d.lastSeq = ev.Seq
|
|
return h, nil
|
|
}
|
|
|
|
// Next returns the next event, or io.EOF at the end of the stream. An unknown event type is
|
|
// returned as-is: tolerating it is the minor-version rule, and dropping it is the sink's decision.
|
|
func (d *Decoder) Next() (Envelope, error) {
|
|
if !d.greeted {
|
|
return Envelope{}, ErrNoHandshake
|
|
}
|
|
ev, err := d.line()
|
|
if err != nil {
|
|
return Envelope{}, err
|
|
}
|
|
if ev.Type == TypeHello {
|
|
return Envelope{}, fmt.Errorf("%w: seq %d", ErrRepeatedHello, ev.Seq)
|
|
}
|
|
// Exactly one step, in one direction. A repeat is as wrong as a gap: inside one pipe there is
|
|
// no at-least-once redelivery to absorb, so both are reported.
|
|
if ev.Seq != d.lastSeq+1 {
|
|
return Envelope{}, fmt.Errorf("%w: seq %d after %d", ErrStreamGap, ev.Seq, d.lastSeq)
|
|
}
|
|
d.lastSeq = ev.Seq
|
|
return ev, nil
|
|
}
|
|
|
|
func (d *Decoder) line() (Envelope, error) {
|
|
for d.sc.Scan() {
|
|
raw := d.sc.Bytes()
|
|
if len(raw) == 0 {
|
|
continue // a blank line is not an event
|
|
}
|
|
var ev Envelope
|
|
if err := json.Unmarshal(raw, &ev); err != nil {
|
|
return Envelope{}, fmt.Errorf("ingest: malformed line: %w", err)
|
|
}
|
|
return ev, nil
|
|
}
|
|
if err := d.sc.Err(); err != nil {
|
|
return Envelope{}, fmt.Errorf("ingest: read stream: %w", err)
|
|
}
|
|
return Envelope{}, io.EOF
|
|
}
|