| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/binary" |
| 7 | "encoding/hex" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "hash" |
| 12 | "io" |
| 13 | "strings" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/klauspost/compress/zstd" |
| 17 | |
| 18 | "reasonix/internal/sessioncontent" |
| 19 | ) |
| 20 | |
| 21 | const ( |
| 22 | V4SchemaVersion = 4 |
| 23 | V4Codec = "reasonix.session.linear/v4" |
| 24 | v4InlinePayloadBytes = 64 << 10 |
| 25 | v4MaxFrameBytes = 8 << 20 |
| 26 | v4FrameHeaderBytes = 12 |
| 27 | ) |
| 28 | |
| 29 | var v4FrameMagic = [4]byte{'R', 'X', '4', 'F'} |
| 30 | |
| 31 | type v4Record struct { |
| 32 | SchemaVersion int `json:"schemaVersion"` |
| 33 | Codec string `json:"codec"` |
| 34 | RecordType string `json:"recordType"` |
| 35 | |
| 36 | CommitID string `json:"commitId,omitempty"` |
| 37 | OperationID string `json:"operationId,omitempty"` |
| 38 | OperationHash string `json:"operationHash,omitempty"` |
| 39 | FirstSequence uint64 `json:"firstSeq,omitempty"` |
| 40 | EventCount int `json:"eventCount,omitempty"` |
| 41 | TurnID string `json:"turnId,omitempty"` |
| 42 | WriterGeneration uint64 `json:"writerGeneration,omitempty"` |
| 43 | CreatedAt time.Time `json:"createdAt,omitempty"` |
| 44 | |
| 45 | Event *v4Event `json:"event,omitempty"` |
| 46 | SHA256 string `json:"sha256,omitempty"` |
| 47 | } |
| 48 | |
| 49 | type v4Event struct { |
| 50 | ID string `json:"id"` |
| 51 | Sequence uint64 `json:"seq"` |
| 52 | Kind string `json:"kind"` |
| 53 | Optional bool `json:"optional,omitempty"` |
| 54 | Required bool `json:"required,omitempty"` |
| 55 | Payload []byte `json:"payload,omitempty"` |
| 56 | PayloadRef *sessioncontent.Ref `json:"payloadRef,omitempty"` |
| 57 | } |
| 58 | |
| 59 | // encodeV4Commits writes each logical commit as an atomic begin/event/end |
| 60 | // transaction. Every record is an independent checksummed Zstandard frame. |
| 61 | // Large payload bytes are durably published to content before their reference |
| 62 | // can enter the log. |
| 63 | func encodeV4Commits(ctx context.Context, dst io.Writer, content *sessioncontent.Store, commits []Commit) ([]int64, error) { |
| 64 | if err := ctx.Err(); err != nil { |
| 65 | return nil, err |
| 66 | } |
| 67 | encoder, err := zstd.NewWriter(nil, |
| 68 | zstd.WithEncoderLevel(zstd.SpeedFastest), |
| 69 | zstd.WithWindowSize(1<<20), |
| 70 | zstd.WithEncoderCRC(true), |
| 71 | zstd.WithEncoderConcurrency(1), |
| 72 | ) |
| 73 | if err != nil { |
| 74 | return nil, fmt.Errorf("sessionv4: create encoder: %w", err) |
| 75 | } |
| 76 | defer encoder.Close() |
| 77 | lengths := make([]int64, 0, len(commits)) |
| 78 | for _, commit := range commits { |
| 79 | if err := ctx.Err(); err != nil { |
| 80 | return nil, err |
| 81 | } |
| 82 | counting := &countingWriter{writer: dst} |
| 83 | digest := sha256.New() |
| 84 | begin := v4Record{ |
| 85 | SchemaVersion: V4SchemaVersion, Codec: V4Codec, RecordType: "batch/begin", |
| 86 | CommitID: commit.ID, OperationID: commit.OperationID, OperationHash: commit.OperationHash, |
| 87 | FirstSequence: commit.FirstSequence, EventCount: commit.EventCount, TurnID: commit.TurnID, |
| 88 | WriterGeneration: commit.WriterGeneration, CreatedAt: commit.CreatedAt, |
| 89 | } |
| 90 | if err := writeV4DigestRecord(ctx, counting, encoder, digest, begin); err != nil { |
| 91 | return nil, err |
| 92 | } |
| 93 | for _, event := range commit.Events { |
| 94 | recordEvent := &v4Event{ |
| 95 | ID: event.ID, Sequence: event.Sequence, Kind: event.Kind, |
| 96 | Optional: event.Optional, Required: event.Required, |
| 97 | } |
| 98 | if event.PayloadRef != nil { |
| 99 | if len(event.Payload) != 0 || content == nil { |
| 100 | return nil, errors.New("sessionv4: referenced payload must have one available content store") |
| 101 | } |
| 102 | if err := content.Verify(ctx, *event.PayloadRef); err != nil { |
| 103 | return nil, fmt.Errorf("sessionv4: verify event %s payload: %w", event.ID, err) |
| 104 | } |
| 105 | ref := *event.PayloadRef |
| 106 | recordEvent.PayloadRef = &ref |
| 107 | } else if len(event.Payload) > v4InlinePayloadBytes { |
| 108 | if content == nil { |
| 109 | return nil, errors.New("sessionv4: content store is required for a large payload") |
| 110 | } |
| 111 | ref, putErr := content.Put(ctx, bytesReader(event.Payload), sessioncontent.Metadata{MediaType: "application/json"}) |
| 112 | if putErr != nil { |
| 113 | return nil, fmt.Errorf("sessionv4: store event %s payload: %w", event.ID, putErr) |
| 114 | } |
| 115 | recordEvent.PayloadRef = &ref |
| 116 | } else { |
| 117 | recordEvent.Payload = append([]byte(nil), event.Payload...) |
| 118 | } |
| 119 | if err := writeV4DigestRecord(ctx, counting, encoder, digest, v4Record{ |
| 120 | SchemaVersion: V4SchemaVersion, Codec: V4Codec, RecordType: "batch/event", Event: recordEvent, |
| 121 | }); err != nil { |
| 122 | return nil, err |
| 123 | } |
| 124 | } |
| 125 | end := v4Record{ |
| 126 | SchemaVersion: V4SchemaVersion, Codec: V4Codec, RecordType: "batch/end", |
| 127 | CommitID: commit.ID, FirstSequence: commit.FirstSequence, EventCount: commit.EventCount, |
| 128 | SHA256: hex.EncodeToString(digest.Sum(nil)), |
| 129 | } |
| 130 | if err := writeV4Record(ctx, counting, encoder, end); err != nil { |
| 131 | return nil, err |
| 132 | } |
| 133 | lengths = append(lengths, counting.count) |
| 134 | } |
| 135 | return lengths, nil |
| 136 | } |
| 137 | |
| 138 | func writeV4DigestRecord(ctx context.Context, dst io.Writer, encoder *zstd.Encoder, digest hash.Hash, record v4Record) error { |
| 139 | raw, err := json.Marshal(record) |
| 140 | if err != nil { |
| 141 | return err |
| 142 | } |
| 143 | if _, err := digest.Write(raw); err != nil { |
| 144 | return err |
| 145 | } |
| 146 | if _, err := digest.Write([]byte{0}); err != nil { |
| 147 | return err |
| 148 | } |
| 149 | return writeV4RawRecord(ctx, dst, encoder, raw) |
| 150 | } |
| 151 | |
| 152 | func writeV4Record(ctx context.Context, dst io.Writer, encoder *zstd.Encoder, record v4Record) error { |
| 153 | raw, err := json.Marshal(record) |
| 154 | if err != nil { |
| 155 | return err |
| 156 | } |
| 157 | return writeV4RawRecord(ctx, dst, encoder, raw) |
| 158 | } |
| 159 | |
| 160 | func writeV4RawRecord(ctx context.Context, dst io.Writer, encoder *zstd.Encoder, raw []byte) error { |
| 161 | if err := ctx.Err(); err != nil { |
| 162 | return err |
| 163 | } |
| 164 | if len(raw) == 0 || len(raw) > v4MaxFrameBytes { |
| 165 | return fmt.Errorf("sessionv4: physical record size %d exceeds frame budget %d", len(raw), v4MaxFrameBytes) |
| 166 | } |
| 167 | compressed := encoder.EncodeAll(raw, nil) |
| 168 | if len(compressed) == 0 || len(compressed) > v4MaxFrameBytes { |
| 169 | return fmt.Errorf("sessionv4: compressed frame size %d exceeds frame budget %d", len(compressed), v4MaxFrameBytes) |
| 170 | } |
| 171 | var header [v4FrameHeaderBytes]byte |
| 172 | copy(header[:4], v4FrameMagic[:]) |
| 173 | binary.BigEndian.PutUint32(header[4:8], uint32(len(compressed))) |
| 174 | binary.BigEndian.PutUint32(header[8:12], uint32(len(raw))) |
| 175 | if err := writeAllContext(ctx, dst, header[:]); err != nil { |
| 176 | return err |
| 177 | } |
| 178 | return writeAllContext(ctx, dst, compressed) |
| 179 | } |
| 180 | |
| 181 | // scanV4CommitFile validates transaction framing and exposes only batches |
| 182 | // whose end record and digest are complete. A partial final frame or final |
| 183 | // batch is an uncommitted tail and therefore invisible. |
| 184 | func scanV4CommitFile(ctx context.Context, file io.ReadSeeker, startOffset int64, nextSequence uint64, content *sessioncontent.Store, knownKinds map[string]bool, visit func(int64, Commit) bool) error { |
| 185 | return scanV4CommitFileMode(ctx, file, startOffset, nextSequence, content, knownKinds, true, visit) |
| 186 | } |
| 187 | |
| 188 | // scanV4CommitFileRefs validates the same durable transaction stream while |
| 189 | // leaving external payloads as references. Index rebuilds and history queries |
| 190 | // use this path so cumulative history is never materialized merely to locate |
| 191 | // records. |
| 192 | func scanV4CommitFileRefs(ctx context.Context, file io.ReadSeeker, startOffset int64, nextSequence uint64, content *sessioncontent.Store, knownKinds map[string]bool, visit func(int64, Commit) bool) error { |
| 193 | return scanV4CommitFileMode(ctx, file, startOffset, nextSequence, content, knownKinds, false, visit) |
| 194 | } |
| 195 | |
| 196 | func scanV4CommitFileMode(ctx context.Context, file io.ReadSeeker, startOffset int64, nextSequence uint64, content *sessioncontent.Store, knownKinds map[string]bool, resolvePayloads bool, visit func(int64, Commit) bool) error { |
| 197 | if knownKinds == nil { |
| 198 | knownKinds = ProjectionKinds |
| 199 | } |
| 200 | if _, err := file.Seek(startOffset, io.SeekStart); err != nil { |
| 201 | return err |
| 202 | } |
| 203 | decoder, err := zstd.NewReader(nil, |
| 204 | zstd.WithDecoderConcurrency(1), |
| 205 | zstd.WithDecoderMaxMemory(v4MaxFrameBytes), |
| 206 | ) |
| 207 | if err != nil { |
| 208 | return fmt.Errorf("sessionv4: create decoder: %w", err) |
| 209 | } |
| 210 | defer decoder.Close() |
| 211 | |
| 212 | var pending *Commit |
| 213 | var pendingOffset int64 |
| 214 | var digest hash.Hash |
| 215 | offset := startOffset |
| 216 | operations := map[string]string{} |
| 217 | for { |
| 218 | recordOffset := offset |
| 219 | raw, frameBytes, complete, readErr := readV4Frame(ctx, file, decoder) |
| 220 | if readErr != nil { |
| 221 | return readErr |
| 222 | } |
| 223 | if !complete { |
| 224 | return nil |
| 225 | } |
| 226 | offset += frameBytes |
| 227 | var record v4Record |
| 228 | if err := json.Unmarshal(raw, &record); err != nil { |
| 229 | return fmt.Errorf("%w: decode v4 record at %d: %w", ErrDamagedStore, recordOffset, err) |
| 230 | } |
| 231 | if record.SchemaVersion != V4SchemaVersion || record.Codec != V4Codec { |
| 232 | return fmt.Errorf("%w: v4 physical record at %d", ErrUnsupportedVersion, recordOffset) |
| 233 | } |
| 234 | switch record.RecordType { |
| 235 | case "batch/begin": |
| 236 | if pending != nil { |
| 237 | return fmt.Errorf("%w: nested v4 batch at %d", ErrDamagedStore, recordOffset) |
| 238 | } |
| 239 | if record.CommitID == "" || record.OperationID == "" || record.OperationHash == "" || record.WriterGeneration == 0 || record.FirstSequence != nextSequence || record.EventCount <= 0 { |
| 240 | return fmt.Errorf("%w: invalid v4 batch boundary at sequence %d", ErrDamagedStore, nextSequence) |
| 241 | } |
| 242 | pending = &Commit{ |
| 243 | SchemaVersion: V4SchemaVersion, Codec: V4Codec, RecordType: "commit", |
| 244 | ID: record.CommitID, OperationID: record.OperationID, OperationHash: record.OperationHash, |
| 245 | FirstSequence: record.FirstSequence, EventCount: record.EventCount, TurnID: record.TurnID, |
| 246 | WriterGeneration: record.WriterGeneration, CreatedAt: record.CreatedAt, |
| 247 | // Never trust an on-disk cumulative count as an allocation request. |
| 248 | // Capacity grows only as individually bounded frames validate. |
| 249 | Events: nil, |
| 250 | } |
| 251 | pendingOffset = recordOffset |
| 252 | digest = sha256.New() |
| 253 | _, _ = digest.Write(raw) |
| 254 | _, _ = digest.Write([]byte{0}) |
| 255 | case "batch/event": |
| 256 | if pending == nil || record.Event == nil || len(pending.Events) >= pending.EventCount { |
| 257 | return fmt.Errorf("%w: v4 event outside batch at %d", ErrDamagedStore, recordOffset) |
| 258 | } |
| 259 | physical := record.Event |
| 260 | wantSequence := pending.FirstSequence + uint64(len(pending.Events)) |
| 261 | if physical.ID == "" || strings.TrimSpace(physical.Kind) == "" || physical.Sequence != wantSequence || (physical.PayloadRef != nil && physical.Payload != nil) { |
| 262 | return fmt.Errorf("%w: invalid v4 event at sequence %d", ErrDamagedStore, wantSequence) |
| 263 | } |
| 264 | payload := append(json.RawMessage(nil), physical.Payload...) |
| 265 | var payloadRef *sessioncontent.Ref |
| 266 | if physical.PayloadRef != nil { |
| 267 | ref := *physical.PayloadRef |
| 268 | payloadRef = &ref |
| 269 | if resolvePayloads { |
| 270 | payload, err = resolveContentPayload(ctx, content, ref) |
| 271 | if err != nil { |
| 272 | return fmt.Errorf("%w: read v4 event %s payload: %w", ErrDamagedStore, physical.ID, err) |
| 273 | } |
| 274 | payloadRef = nil |
| 275 | } |
| 276 | } |
| 277 | pending.Events = append(pending.Events, Event{ |
| 278 | ID: physical.ID, Sequence: physical.Sequence, Kind: physical.Kind, |
| 279 | Optional: physical.Optional, Required: physical.Required, Payload: payload, PayloadRef: payloadRef, |
| 280 | }) |
| 281 | _, _ = digest.Write(raw) |
| 282 | _, _ = digest.Write([]byte{0}) |
| 283 | case "batch/end": |
| 284 | if pending == nil || record.CommitID != pending.ID || record.FirstSequence != pending.FirstSequence || record.EventCount != pending.EventCount || len(pending.Events) != pending.EventCount { |
| 285 | return fmt.Errorf("%w: invalid v4 batch end at %d", ErrDamagedStore, recordOffset) |
| 286 | } |
| 287 | if got := hex.EncodeToString(digest.Sum(nil)); record.SHA256 != got { |
| 288 | return fmt.Errorf("%w: v4 batch %s checksum is %s, expected %s", ErrDamagedStore, pending.ID, record.SHA256, got) |
| 289 | } |
| 290 | if prior, ok := operations[pending.OperationID]; ok && prior != pending.OperationHash { |
| 291 | return fmt.Errorf("%w: conflicting operation %q", ErrDamagedStore, pending.OperationID) |
| 292 | } |
| 293 | operations[pending.OperationID] = pending.OperationHash |
| 294 | for _, event := range pending.Events { |
| 295 | if !event.Optional && !knownKinds[event.Kind] { |
| 296 | return fmt.Errorf("%w: unknown required event %q", ErrUnsupportedVersion, event.Kind) |
| 297 | } |
| 298 | } |
| 299 | nextSequence = pending.LastSequence() + 1 |
| 300 | completed := *pending |
| 301 | pending = nil |
| 302 | digest = nil |
| 303 | if visit != nil && !visit(pendingOffset, completed) { |
| 304 | return nil |
| 305 | } |
| 306 | default: |
| 307 | return fmt.Errorf("%w: unknown v4 physical record %q", ErrUnsupportedVersion, record.RecordType) |
| 308 | } |
| 309 | } |
| 310 | } |
| 311 | |
| 312 | func resolveContentPayload(ctx context.Context, content *sessioncontent.Store, ref sessioncontent.Ref) (json.RawMessage, error) { |
| 313 | if content == nil { |
| 314 | return nil, errors.New("sessionv4: content store is required to resolve a payload reference") |
| 315 | } |
| 316 | if ref.Bytes < 0 || uint64(ref.Bytes) > uint64(^uint(0)>>1) { |
| 317 | return nil, fmt.Errorf("sessionv4: payload size %d cannot be materialized by this process", ref.Bytes) |
| 318 | } |
| 319 | r, err := content.Open(ctx, ref) |
| 320 | if err != nil { |
| 321 | return nil, err |
| 322 | } |
| 323 | defer r.Close() |
| 324 | payload := make([]byte, int(ref.Bytes)) |
| 325 | if _, err := io.ReadFull(r, payload); err != nil { |
| 326 | return nil, err |
| 327 | } |
| 328 | return payload, nil |
| 329 | } |
| 330 | |
| 331 | func readV4Frame(ctx context.Context, reader io.Reader, decoder *zstd.Decoder) ([]byte, int64, bool, error) { |
| 332 | if err := ctx.Err(); err != nil { |
| 333 | return nil, 0, false, err |
| 334 | } |
| 335 | var header [v4FrameHeaderBytes]byte |
| 336 | n, err := io.ReadFull(reader, header[:]) |
| 337 | if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { |
| 338 | return nil, int64(n), false, nil |
| 339 | } |
| 340 | if err != nil { |
| 341 | return nil, int64(n), false, err |
| 342 | } |
| 343 | if [4]byte(header[:4]) != v4FrameMagic { |
| 344 | return nil, 0, false, fmt.Errorf("%w: invalid v4 frame magic", ErrDamagedStore) |
| 345 | } |
| 346 | compressedBytes := int(binary.BigEndian.Uint32(header[4:8])) |
| 347 | rawBytes := int(binary.BigEndian.Uint32(header[8:12])) |
| 348 | if compressedBytes <= 0 || compressedBytes > v4MaxFrameBytes || rawBytes <= 0 || rawBytes > v4MaxFrameBytes { |
| 349 | return nil, 0, false, fmt.Errorf("%w: invalid v4 frame sizes compressed=%d raw=%d", ErrDamagedStore, compressedBytes, rawBytes) |
| 350 | } |
| 351 | compressed := make([]byte, compressedBytes) |
| 352 | n, err = io.ReadFull(reader, compressed) |
| 353 | if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { |
| 354 | return nil, int64(v4FrameHeaderBytes + n), false, nil |
| 355 | } |
| 356 | if err != nil { |
| 357 | return nil, int64(v4FrameHeaderBytes + n), false, err |
| 358 | } |
| 359 | if err := ctx.Err(); err != nil { |
| 360 | return nil, 0, false, err |
| 361 | } |
| 362 | raw, err := decoder.DecodeAll(compressed, make([]byte, 0, rawBytes)) |
| 363 | if err != nil { |
| 364 | return nil, 0, false, fmt.Errorf("%w: decode v4 frame: %w", ErrDamagedStore, err) |
| 365 | } |
| 366 | if len(raw) != rawBytes { |
| 367 | return nil, 0, false, fmt.Errorf("%w: v4 frame decoded %d bytes, expected %d", ErrDamagedStore, len(raw), rawBytes) |
| 368 | } |
| 369 | return raw, int64(v4FrameHeaderBytes + compressedBytes), true, nil |
| 370 | } |
| 371 | |
| 372 | type countingWriter struct { |
| 373 | writer io.Writer |
| 374 | count int64 |
| 375 | } |
| 376 | |
| 377 | func (w *countingWriter) Write(p []byte) (int, error) { |
| 378 | n, err := w.writer.Write(p) |
| 379 | w.count += int64(n) |
| 380 | return n, err |
| 381 | } |
| 382 | |
| 383 | type rawBytesReader struct { |
| 384 | data []byte |
| 385 | off int |
| 386 | } |
| 387 | |
| 388 | func bytesReader(data []byte) *rawBytesReader { return &rawBytesReader{data: data} } |
| 389 | |
| 390 | func (r *rawBytesReader) Read(p []byte) (int, error) { |
| 391 | if r.off >= len(r.data) { |
| 392 | return 0, io.EOF |
| 393 | } |
| 394 | n := copy(p, r.data[r.off:]) |
| 395 | r.off += n |
| 396 | return n, nil |
| 397 | } |
| 398 |