98 lines
3.5 KiB
Go
98 lines
3.5 KiB
Go
package ingest
|
|
|
|
import (
|
|
"errors"
|
|
"io"
|
|
"strings"
|
|
"testing"
|
|
)
|
|
|
|
// The decoder is the only place where bytes produced by another process become platform state, so
|
|
// it is fuzzed rather than merely exampled. The oracles are the invariants the rest of the ingest
|
|
// path is built on; each is stated as "if the decoder said yes, then …", because a refusal is
|
|
// always an acceptable answer and only an ACCEPTANCE can be wrong.
|
|
//
|
|
// 1. it never panics, whatever arrives;
|
|
// 2. an accepted handshake identifies the stream (engine_run_id non-empty — half of the
|
|
// idempotency key) and is seq 1;
|
|
// 3. accepted events increase seq by exactly one, so a gap can never be silently absorbed;
|
|
// 4. no event is ever returned before a successful handshake;
|
|
// 5. hello never appears again after the handshake, at any version.
|
|
func FuzzDecoder(f *testing.F) {
|
|
f.Add(helloLine)
|
|
f.Add(helloLine + "\n" + `{"seq":2,"type":"progress","data":{"draft":{"done":1,"total":2}}}`)
|
|
f.Add(helloLine + "\n" + helloLine)
|
|
f.Add(`{"seq":9,"type":"hello","data":{"stream_version":"1.0","engine_run_id":"x"}}`)
|
|
f.Add(`{"seq":1,"type":"hello","data":{"stream_version":"1.0","engine_run_id":""}}`)
|
|
f.Add(`{"seq":1,"type":"hello","data":{"stream_version":"9.9","engine_run_id":"x"}}`)
|
|
f.Add("\n\n\n")
|
|
f.Add("{not json")
|
|
|
|
f.Fuzz(func(t *testing.T, stream string) {
|
|
d := NewDecoder(strings.NewReader(stream))
|
|
|
|
// Oracle 4: nothing may come out before the handshake.
|
|
if _, err := d.Next(); !errors.Is(err, ErrNoHandshake) {
|
|
t.Fatalf("Next before Hello returned %v, want ErrNoHandshake", err)
|
|
}
|
|
|
|
h, err := d.Hello()
|
|
if err != nil {
|
|
return // a refusal is always allowed
|
|
}
|
|
// Oracle 2.
|
|
if h.EngineRunID == "" {
|
|
t.Fatal("accepted a handshake with no engine_run_id: half the idempotency key")
|
|
}
|
|
if maj, err := major(h.StreamVersion); err != nil || maj != 1 {
|
|
t.Fatalf("accepted stream version %q", h.StreamVersion)
|
|
}
|
|
|
|
last := int64(1) // oracle 2: the handshake is seq 1 or it is refused
|
|
for {
|
|
ev, err := d.Next()
|
|
if err != nil {
|
|
if errors.Is(err, io.EOF) {
|
|
return
|
|
}
|
|
return // any refusal is allowed; only acceptances are constrained
|
|
}
|
|
// Oracle 3.
|
|
if ev.Seq != last+1 {
|
|
t.Fatalf("accepted seq %d after %d", ev.Seq, last)
|
|
}
|
|
// Oracle 5.
|
|
if ev.Type == TypeHello {
|
|
t.Fatal("accepted a second handshake mid-stream")
|
|
}
|
|
last = ev.Seq
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestHandshakeMustIdentifyTheStream(t *testing.T) {
|
|
t.Run("no engine_run_id", func(t *testing.T) {
|
|
line := strings.Replace(helloLine, `"engine_run_id":"tr_1"`, `"engine_run_id":""`, 1)
|
|
if _, err := NewDecoder(strings.NewReader(line)).Hello(); !errors.Is(err, ErrBadHandshake) {
|
|
t.Fatalf("want ErrBadHandshake, got %v", err)
|
|
}
|
|
})
|
|
t.Run("handshake is seq 1", func(t *testing.T) {
|
|
line := strings.Replace(helloLine, `"seq":1`, `"seq":4`, 1)
|
|
if _, err := NewDecoder(strings.NewReader(line)).Hello(); !errors.Is(err, ErrStreamGap) {
|
|
t.Fatalf("want ErrStreamGap, got %v", err)
|
|
}
|
|
})
|
|
t.Run("no second handshake", func(t *testing.T) {
|
|
second := strings.Replace(helloLine, `"seq":1`, `"seq":2`, 1)
|
|
// A major version the gate would have refused on line 1, arriving on line 2.
|
|
second = strings.Replace(second, `"stream_version":"1.0"`, `"stream_version":"9.9"`, 1)
|
|
d := NewDecoder(strings.NewReader(helloLine + "\n" + second))
|
|
if _, err := d.Hello(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := d.Next(); !errors.Is(err, ErrRepeatedHello) {
|
|
t.Fatalf("want ErrRepeatedHello, got %v", err)
|
|
}
|
|
})
|
|
}
|