103 lines
3.1 KiB
Go
103 lines
3.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")
|
|
)
|
|
|
|
// 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)
|
|
}
|
|
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
|
|
}
|
|
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
|
|
}
|
|
switch {
|
|
case ev.Seq <= d.lastSeq:
|
|
// A duplicate cannot happen inside one pipe, so it is a defect rather than at-least-once
|
|
// redelivery — and it is reported, not silently absorbed.
|
|
return Envelope{}, fmt.Errorf("%w: seq %d after %d", ErrStreamGap, ev.Seq, d.lastSeq)
|
|
case 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
|
|
}
|