| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/binary" |
| 7 | "encoding/json" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/klauspost/compress/zstd" |
| 14 | |
| 15 | "reasonix/internal/sessioncontent" |
| 16 | ) |
| 17 | |
| 18 | func TestV4CodecExternalizesLargePayloadAndRoundTripsExactBytes(t *testing.T) { |
| 19 | t.Parallel() |
| 20 | dir := t.TempDir() |
| 21 | content := sessioncontent.New(filepath.Join(dir, "content")) |
| 22 | payload := append([]byte(`{"text":"`), bytes.Repeat([]byte("x"), v4InlinePayloadBytes+12345)...) |
| 23 | payload = append(payload, []byte(`"}`)...) |
| 24 | commit := v4TestCommit(payload) |
| 25 | var log bytes.Buffer |
| 26 | lengths, err := encodeV4Commits(context.Background(), &log, content, []Commit{commit}) |
| 27 | if err != nil { |
| 28 | t.Fatalf("encodeV4Commits: %v", err) |
| 29 | } |
| 30 | if len(lengths) != 1 || lengths[0] != int64(log.Len()) { |
| 31 | t.Fatalf("lengths = %v, log bytes = %d", lengths, log.Len()) |
| 32 | } |
| 33 | for encoded := log.Bytes(); len(encoded) > 0; { |
| 34 | if len(encoded) < v4FrameHeaderBytes { |
| 35 | t.Fatalf("short physical frame header: %d", len(encoded)) |
| 36 | } |
| 37 | compressed := int(binary.BigEndian.Uint32(encoded[4:8])) |
| 38 | raw := int(binary.BigEndian.Uint32(encoded[8:12])) |
| 39 | if compressed > v4MaxFrameBytes || raw > v4MaxFrameBytes { |
| 40 | t.Fatalf("frame exceeds reader budget: compressed=%d raw=%d", compressed, raw) |
| 41 | } |
| 42 | encoded = encoded[v4FrameHeaderBytes+compressed:] |
| 43 | } |
| 44 | |
| 45 | path := filepath.Join(dir, "events.v4") |
| 46 | if err := os.WriteFile(path, log.Bytes(), 0o600); err != nil { |
| 47 | t.Fatalf("write log: %v", err) |
| 48 | } |
| 49 | f, err := os.Open(path) |
| 50 | if err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | defer f.Close() |
| 54 | var got []Commit |
| 55 | err = scanV4CommitFile(context.Background(), f, 0, 1, content, nil, func(_ int64, commit Commit) bool { |
| 56 | got = append(got, commit) |
| 57 | return true |
| 58 | }) |
| 59 | if err != nil { |
| 60 | t.Fatalf("scanV4CommitFile: %v", err) |
| 61 | } |
| 62 | if len(got) != 1 || len(got[0].Events) != 1 || !bytes.Equal(got[0].Events[0].Payload, payload) { |
| 63 | t.Fatalf("round trip mismatch: commits=%d payload=%d", len(got), payloadLen(got)) |
| 64 | } |
| 65 | |
| 66 | entries, err := filepath.Glob(filepath.Join(content.Root(), "objects", "*", "*", "*")) |
| 67 | if err != nil || len(entries) != 1 { |
| 68 | t.Fatalf("external objects = %v, err=%v", entries, err) |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | func TestV4ReferenceScanDoesNotMaterializeLargePayload(t *testing.T) { |
| 73 | t.Parallel() |
| 74 | dir := t.TempDir() |
| 75 | content := sessioncontent.New(filepath.Join(dir, "content")) |
| 76 | payload := bytes.Repeat([]byte("p"), 2*v4InlinePayloadBytes) |
| 77 | commit := v4TestCommit(payload) |
| 78 | var log bytes.Buffer |
| 79 | if _, err := encodeV4Commits(t.Context(), &log, content, []Commit{commit}); err != nil { |
| 80 | t.Fatal(err) |
| 81 | } |
| 82 | file := writeAndOpenV4TestLog(t, dir, "lazy.v4", log.Bytes()) |
| 83 | defer file.Close() |
| 84 | if err := scanV4CommitFileRefs(t.Context(), file, 0, 1, content, nil, func(_ int64, got Commit) bool { |
| 85 | if len(got.Events) != 1 || len(got.Events[0].Payload) != 0 || got.Events[0].PayloadRef == nil || got.Events[0].PayloadRef.Bytes != int64(len(payload)) { |
| 86 | t.Fatalf("lazy event = %#v", got.Events) |
| 87 | } |
| 88 | return true |
| 89 | }); err != nil { |
| 90 | t.Fatal(err) |
| 91 | } |
| 92 | } |
| 93 | |
| 94 | func TestV4ReaderDoesNotPreallocateUntrustedEventCount(t *testing.T) { |
| 95 | t.Parallel() |
| 96 | record := v4Record{ |
| 97 | SchemaVersion: V4SchemaVersion, Codec: V4Codec, RecordType: "batch/begin", |
| 98 | CommitID: "c", OperationID: "o", OperationHash: "h", FirstSequence: 1, |
| 99 | EventCount: int(^uint(0) >> 2), WriterGeneration: 1, |
| 100 | } |
| 101 | encoder, err := zstd.NewWriter(nil, zstd.WithEncoderConcurrency(1)) |
| 102 | if err != nil { |
| 103 | t.Fatal(err) |
| 104 | } |
| 105 | defer encoder.Close() |
| 106 | var log bytes.Buffer |
| 107 | if err := writeV4Record(t.Context(), &log, encoder, record); err != nil { |
| 108 | t.Fatal(err) |
| 109 | } |
| 110 | file := writeAndOpenV4TestLog(t, t.TempDir(), "count.v4", log.Bytes()) |
| 111 | defer file.Close() |
| 112 | if err := scanV4CommitFileRefs(t.Context(), file, 0, 1, nil, nil, nil); err != nil { |
| 113 | t.Fatalf("incomplete declared transaction should remain invisible: %v", err) |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | func TestV4CodecHidesIncompleteBatchAndRejectsCorruption(t *testing.T) { |
| 118 | t.Parallel() |
| 119 | dir := t.TempDir() |
| 120 | content := sessioncontent.New(filepath.Join(dir, "content")) |
| 121 | commit := v4TestCommit(json.RawMessage(`{"message":"durable only after end"}`)) |
| 122 | var log bytes.Buffer |
| 123 | if _, err := encodeV4Commits(context.Background(), &log, content, []Commit{commit}); err != nil { |
| 124 | t.Fatalf("encode: %v", err) |
| 125 | } |
| 126 | |
| 127 | truncated := append([]byte(nil), log.Bytes()[:log.Len()-5]...) |
| 128 | f1 := writeAndOpenV4TestLog(t, dir, "truncated.v4", truncated) |
| 129 | visited := 0 |
| 130 | if err := scanV4CommitFile(context.Background(), f1, 0, 1, content, nil, func(_ int64, _ Commit) bool { |
| 131 | visited++ |
| 132 | return true |
| 133 | }); err != nil { |
| 134 | t.Fatalf("incomplete tail should be ignored, got %v", err) |
| 135 | } |
| 136 | _ = f1.Close() |
| 137 | if visited != 0 { |
| 138 | t.Fatalf("incomplete transaction became visible: %d commits", visited) |
| 139 | } |
| 140 | |
| 141 | corrupt := append([]byte(nil), log.Bytes()...) |
| 142 | if len(corrupt) < v4FrameHeaderBytes+8 { |
| 143 | t.Fatal("encoded test log unexpectedly small") |
| 144 | } |
| 145 | corrupt[v4FrameHeaderBytes+4] ^= 0x7f |
| 146 | f2 := writeAndOpenV4TestLog(t, dir, "corrupt.v4", corrupt) |
| 147 | defer f2.Close() |
| 148 | if err := scanV4CommitFile(context.Background(), f2, 0, 1, content, nil, nil); err == nil { |
| 149 | t.Fatal("corrupt complete frame was accepted") |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | func v4TestCommit(payload json.RawMessage) Commit { |
| 154 | return Commit{ |
| 155 | SchemaVersion: 4, |
| 156 | Codec: V4Codec, |
| 157 | RecordType: "commit", |
| 158 | ID: "commit-1", |
| 159 | OperationID: "operation-1", |
| 160 | OperationHash: "hash-1", |
| 161 | FirstSequence: 1, |
| 162 | EventCount: 1, |
| 163 | TurnID: "turn-1", |
| 164 | WriterGeneration: 1, |
| 165 | CreatedAt: time.Unix(100, 0).UTC(), |
| 166 | Events: []Event{{ID: "event-1", Sequence: 1, Kind: "diagnostic", Payload: payload}}, |
| 167 | } |
| 168 | } |
| 169 | |
| 170 | func payloadLen(commits []Commit) int { |
| 171 | if len(commits) == 0 || len(commits[0].Events) == 0 { |
| 172 | return 0 |
| 173 | } |
| 174 | return len(commits[0].Events[0].Payload) |
| 175 | } |
| 176 | |
| 177 | func writeAndOpenV4TestLog(t *testing.T, dir, name string, data []byte) *os.File { |
| 178 | t.Helper() |
| 179 | path := filepath.Join(dir, name) |
| 180 | if err := os.WriteFile(path, data, 0o600); err != nil { |
| 181 | t.Fatal(err) |
| 182 | } |
| 183 | f, err := os.Open(path) |
| 184 | if err != nil { |
| 185 | t.Fatal(err) |
| 186 | } |
| 187 | return f |
| 188 | } |
| 189 |