test(snapshot): verify incremental persistence performance
- commit
- 407cac9c01c1e4c16b7633defe4c5b88614ab443
- author
- brice · · 2026-07-22 01:46 UTC
- parent
- e8e2f00
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +256 -169
M internal/usecase/daemon/performance_benchmark_test.go +256 -45
| @@ -14,6 +14,7 @@ import ( | |||
| 14 | 14 | "github.com/bnema/vev/internal/ports" | |
| 15 | 15 | "github.com/bnema/vev/internal/usecase/layout" | |
| 16 | 16 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 17 | + | "github.com/bnema/vev/pkg/renderer" | |
| 17 | 18 | ) | |
| 18 | 19 | ||
| 19 | 20 | func TestCountingSnapshotRepositoryPublishCountsAndRetainsManifest(t *testing.T) { | |
| @@ -22,13 +23,78 @@ func TestCountingSnapshotRepositoryPublishCountsAndRetainsManifest(t *testing.T) | |||
| 22 | 23 | object := []byte{0x01, 0x02} | |
| 23 | 24 | ||
| 24 | 25 | require.NoError(t, repository.Publish(context.Background(), ports.SnapshotPublication{Name: "session", Manifest: payload, Objects: []ports.SnapshotObject{{Data: object}}})) | |
| 25 | - | require.Equal(t, countingSnapshotMetrics{writes: 1, manifestBytes: uint64(len(payload)), objectBytes: uint64(len(object))}, repository.metrics()) | |
| 26 | + | require.Equal(t, countingSnapshotMetrics{writes: 1, manifestBytes: uint64(len(payload)), objectBytes: uint64(len(object)), headBytes: snapshotHeadBytes}, repository.metrics()) | |
| 26 | 27 | require.Equal(t, payload, repository.lastPayload()) | |
| 27 | 28 | ||
| 28 | 29 | payload[0] = 0x01 | |
| 29 | 30 | require.Equal(t, payload, repository.lastPayload(), "the counting sink must retain the payload slice header, not copy it") | |
| 30 | 31 | } | |
| 31 | 32 | ||
| 33 | + | func TestCountingSnapshotRepositoryDoesNotRewriteKnownObjects(t *testing.T) { | |
| 34 | + | repository := &countingSnapshotRepository{} | |
| 35 | + | object, err := snapcodec.MarshalObject(snapcodec.HistoryChunk, []byte{0x01}) | |
| 36 | + | require.NoError(t, err) | |
| 37 | + | publication := ports.SnapshotPublication{Name: "session", Manifest: []byte("manifest"), Objects: []ports.SnapshotObject{object}} | |
| 38 | + | ||
| 39 | + | require.NoError(t, repository.Publish(context.Background(), publication)) | |
| 40 | + | repository.reset() | |
| 41 | + | require.NoError(t, repository.Publish(context.Background(), publication)) | |
| 42 | + | ||
| 43 | + | require.Zero(t, repository.metrics().objectBytes, "filesystem repositories retain immutable content-addressed objects") | |
| 44 | + | } | |
| 45 | + | ||
| 46 | + | func TestIncrementalSnapshotMetricsWriteNoUnchangedHistoryBlobsAndBoundCache(t *testing.T) { | |
| 47 | + | fixture := newPerformanceFixture(t, performanceConfig{size: domain.Size{Cols: 120, Rows: 40}, panes: 1, historyRows: 10_000}) | |
| 48 | + | ||
| 49 | + | publishBenchmarkSnapshot(t, fixture, 1) | |
| 50 | + | fixture.snapshots.reset() | |
| 51 | + | publishBenchmarkSnapshot(t, fixture, 2) | |
| 52 | + | ||
| 53 | + | metrics := fixture.snapshots.metrics() | |
| 54 | + | require.Zero(t, metrics.historyBlobBytes) | |
| 55 | + | require.Zero(t, metrics.objectBytes) | |
| 56 | + | require.Positive(t, metrics.manifestBytes) | |
| 57 | + | require.Equal(t, uint64(snapshotHeadBytes), metrics.headBytes) | |
| 58 | + | require.LessOrEqual(t, benchmarkSnapshotCacheBytes(fixture.sess), snapshotChunkCacheLimit) | |
| 59 | + | } | |
| 60 | + | ||
| 61 | + | func TestIncrementalSnapshotMetricScenarios(t *testing.T) { | |
| 62 | + | for _, tt := range []struct { | |
| 63 | + | name string | |
| 64 | + | historyRows int | |
| 65 | + | mutate func(*performanceFixture, int) | |
| 66 | + | wantHistory bool | |
| 67 | + | }{ | |
| 68 | + | {name: "tail-only", historyRows: 9_999, mutate: mutateBenchmarkTail}, | |
| 69 | + | {name: "new-sealed-chunk", historyRows: 10_000, mutate: mutateBenchmarkSealedChunk, wantHistory: true}, | |
| 70 | + | } { | |
| 71 | + | t.Run(tt.name, func(t *testing.T) { | |
| 72 | + | fixture := newPerformanceFixture(t, performanceConfig{size: domain.Size{Cols: 120, Rows: 40}, panes: 1, historyRows: tt.historyRows}) | |
| 73 | + | publishBenchmarkSnapshot(t, fixture, 1) | |
| 74 | + | fixture.snapshots.reset() | |
| 75 | + | tt.mutate(fixture, 0) | |
| 76 | + | publishBenchmarkSnapshot(t, fixture, 2) | |
| 77 | + | ||
| 78 | + | metrics := fixture.snapshots.metrics() | |
| 79 | + | if tt.wantHistory { | |
| 80 | + | require.Positive(t, metrics.historyBlobBytes) | |
| 81 | + | } else { | |
| 82 | + | require.Zero(t, metrics.historyBlobBytes) | |
| 83 | + | } | |
| 84 | + | require.LessOrEqual(t, benchmarkSnapshotCacheBytes(fixture.sess), snapshotChunkCacheLimit) | |
| 85 | + | }) | |
| 86 | + | } | |
| 87 | + | } | |
| 88 | + | ||
| 89 | + | func TestPerformanceFixtureCellLimitedHistory(t *testing.T) { | |
| 90 | + | fixture := newPerformanceFixture(t, performanceConfig{size: domain.Size{Cols: 293, Rows: 40}, panes: 1, historyRows: 10_000}) | |
| 91 | + | fixture.activePane.mu.Lock() | |
| 92 | + | rows, cells, capCells := fixture.activePane.history.Len(), fixture.activePane.history.Cells(), fixture.activePane.history.CellCap() | |
| 93 | + | fixture.activePane.mu.Unlock() | |
| 94 | + | require.Less(t, rows, 10_000) | |
| 95 | + | require.LessOrEqual(t, cells, capCells) | |
| 96 | + | } | |
| 97 | + | ||
| 32 | 98 | func TestCountingOutputTransportCountsOpaquePayloadAndRejectsShortPayload(t *testing.T) { | |
| 33 | 99 | transport := &countingOutputTransport{} | |
| 34 | 100 | ||
| @@ -219,10 +285,16 @@ func BenchmarkDaemonHistorySnapshotCapture(b *testing.B) { | |||
| 219 | 285 | benchmarkDaemonSnapshotCapture(b) | |
| 220 | 286 | } | |
| 221 | 287 | ||
| 222 | - | // BenchmarkDaemonHistorySnapshotEncode measures incremental publication from an | |
| 223 | - | // already immutable capture, without queueing or persistence. | |
| 224 | - | func BenchmarkDaemonHistorySnapshotEncode(b *testing.B) { | |
| 225 | - | benchmarkDaemonSnapshotEncode(b) | |
| 288 | + | // BenchmarkDaemonIncrementalSnapshotRepository measures the complete | |
| 289 | + | // content-addressed publication path against a filesystem-equivalent byte | |
| 290 | + | // counter. Each scenario reports persisted VEVO, VEVM, and VEVH bytes, plus | |
| 291 | + | // allocation/time and the per-session encoded-history cache high-water mark. | |
| 292 | + | func BenchmarkDaemonIncrementalSnapshotRepository(b *testing.B) { | |
| 293 | + | for _, scenario := range snapshotBenchmarkScenarios { | |
| 294 | + | b.Run(scenario.name, func(b *testing.B) { | |
| 295 | + | benchmarkIncrementalSnapshotRepository(b, scenario) | |
| 296 | + | }) | |
| 297 | + | } | |
| 226 | 298 | } | |
| 227 | 299 | ||
| 228 | 300 | // BenchmarkDaemonHistorySnapshotRestore measures terminal-state restoration | |
| @@ -351,43 +423,156 @@ func benchmarkDaemonSnapshotCapture(b *testing.B) { | |||
| 351 | 423 | } | |
| 352 | 424 | } | |
| 353 | 425 | ||
| 354 | - | func benchmarkDaemonSnapshotEncode(b *testing.B) { | |
| 355 | - | for _, topology := range daemonSnapshotTopologies { | |
| 356 | - | b.Run(topology.name, func(b *testing.B) { | |
| 357 | - | fixture := newPerformanceFixture(b, performanceConfig{size: domain.Size{Cols: 120, Rows: 40}, tabs: topology.tabs, panes: topology.panes, historyRows: 10_000}) | |
| 358 | - | if !fixture.hasHistoryTopology(topology.tabs, topology.panes, 10_000) { | |
| 359 | - | b.Fatal("invalid daemon snapshot publication fixture") | |
| 360 | - | } | |
| 361 | - | capture, ok := fixture.d.captureSnapshotState(fixture.sess, 1) | |
| 362 | - | if !ok || !validBenchmarkCapture(capture) { | |
| 363 | - | b.Fatal("invalid snapshot publication fixture") | |
| 364 | - | } | |
| 365 | - | publication, err := fixture.d.incrementalPublication(capture) | |
| 366 | - | if err != nil { | |
| 367 | - | b.Fatal(err) | |
| 368 | - | } | |
| 369 | - | if _, err := snapcodec.UnmarshalManifest(publication.Manifest); err != nil || len(publication.Objects) == 0 { | |
| 370 | - | b.Fatalf("invalid incremental snapshot publication fixture: %v", err) | |
| 371 | - | } | |
| 372 | - | var last ports.SnapshotPublication | |
| 373 | - | b.ReportAllocs() | |
| 374 | - | b.SetBytes(int64(snapshotPublicationBytes(publication))) | |
| 375 | - | b.ResetTimer() | |
| 376 | - | for range b.N { | |
| 377 | - | last, err = fixture.d.incrementalPublication(capture) | |
| 378 | - | if err != nil { | |
| 379 | - | b.Fatal(err) | |
| 380 | - | } | |
| 381 | - | } | |
| 382 | - | b.StopTimer() | |
| 383 | - | if _, err := snapcodec.UnmarshalManifest(last.Manifest); err != nil { | |
| 384 | - | b.Fatalf("snapshot publication produced invalid manifest: %v", err) | |
| 385 | - | } | |
| 386 | - | b.ReportMetric(float64(len(last.Manifest)), "manifestbytes/op") | |
| 387 | - | b.ReportMetric(float64(snapshotObjectBytes(last)), "objectbytes/op") | |
| 388 | - | b.ReportMetric(float64(len(last.Objects)), "objects/op") | |
| 389 | - | }) | |
| 426 | + | type snapshotBenchmarkScenario struct { | |
| 427 | + | name string | |
| 428 | + | size domain.Size | |
| 429 | + | historyRows int | |
| 430 | + | baseline bool | |
| 431 | + | mutate func(*performanceFixture, int) | |
| 432 | + | unchanged bool | |
| 433 | + | cellLimited bool | |
| 434 | + | } | |
| 435 | + | ||
| 436 | + | var snapshotBenchmarkScenarios = []snapshotBenchmarkScenario{ | |
| 437 | + | {name: "initial-10k-x-120", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 10_000}, | |
| 438 | + | {name: "unchanged", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 10_000, baseline: true, unchanged: true}, | |
| 439 | + | {name: "visible-only", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 10_000, baseline: true, mutate: mutateBenchmarkVisible}, | |
| 440 | + | // Leave one row below the retention cap so the mutation changes only the | |
| 441 | + | // mutable tail and cannot evict a sealed chunk. | |
| 442 | + | {name: "tail-only", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 9_999, baseline: true, mutate: mutateBenchmarkTail}, | |
| 443 | + | {name: "new-sealed-chunk", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 10_000, baseline: true, mutate: mutateBenchmarkSealedChunk}, | |
| 444 | + | {name: "cell-limited-293-columns", size: domain.Size{Cols: 293, Rows: 40}, historyRows: 10_000, cellLimited: true}, | |
| 445 | + | } | |
| 446 | + | ||
| 447 | + | func benchmarkIncrementalSnapshotRepository(b *testing.B, scenario snapshotBenchmarkScenario) { | |
| 448 | + | var ( | |
| 449 | + | metrics countingSnapshotMetrics | |
| 450 | + | peakCacheBytes int | |
| 451 | + | ) | |
| 452 | + | b.ReportAllocs() | |
| 453 | + | b.ResetTimer() | |
| 454 | + | for operation := 0; b.Loop(); operation++ { | |
| 455 | + | // A new fixture is deliberately outside the timed region. It gives every | |
| 456 | + | // iteration the same filesystem state: initial has no generation, | |
| 457 | + | // unchanged/visible/tail/sealed begin after one persisted generation. | |
| 458 | + | // Reusing one fixture would eventually turn a tail into a sealed chunk | |
| 459 | + | // and dilute initial-write bytes over later idempotent publications. | |
| 460 | + | b.StopTimer() | |
| 461 | + | fixture := newPerformanceFixture(b, performanceConfig{size: scenario.size, panes: 1, historyRows: scenario.historyRows}) | |
| 462 | + | validateSnapshotBenchmarkFixture(b, fixture, scenario) | |
| 463 | + | generation := uint64(1) | |
| 464 | + | if scenario.baseline { | |
| 465 | + | publishBenchmarkSnapshot(b, fixture, generation) | |
| 466 | + | generation++ | |
| 467 | + | fixture.snapshots.reset() | |
| 468 | + | } | |
| 469 | + | b.StartTimer() | |
| 470 | + | if scenario.mutate != nil { | |
| 471 | + | scenario.mutate(fixture, operation) | |
| 472 | + | } | |
| 473 | + | publishBenchmarkSnapshot(b, fixture, generation) | |
| 474 | + | b.StopTimer() | |
| 475 | + | ||
| 476 | + | metrics = addCountingSnapshotMetrics(metrics, fixture.snapshots.metrics()) | |
| 477 | + | peakCacheBytes = max(peakCacheBytes, benchmarkSnapshotCacheBytes(fixture.sess)) | |
| 478 | + | // B.Loop requires the timer to be running at the next iteration. | |
| 479 | + | b.StartTimer() | |
| 480 | + | } | |
| 481 | + | b.StopTimer() | |
| 482 | + | if scenario.unchanged && metrics.historyBlobBytes != 0 { | |
| 483 | + | b.Fatalf("unchanged snapshot wrote %d history blob bytes", metrics.historyBlobBytes) | |
| 484 | + | } | |
| 485 | + | if peakCacheBytes > snapshotChunkCacheLimit { | |
| 486 | + | b.Fatalf("snapshot cache peak = %d, limit = %d", peakCacheBytes, snapshotChunkCacheLimit) | |
| 487 | + | } | |
| 488 | + | benchmarkReportSnapshotRepositoryMetrics(b, metrics, b.N, peakCacheBytes) | |
| 489 | + | } | |
| 490 | + | ||
| 491 | + | func validateSnapshotBenchmarkFixture(b *testing.B, fixture *performanceFixture, scenario snapshotBenchmarkScenario) { | |
| 492 | + | b.Helper() | |
| 493 | + | if !fixture.hasHistoryTopology(1, 1, scenario.historyRows) && !scenario.cellLimited { | |
| 494 | + | b.Fatal("invalid incremental snapshot fixture") | |
| 495 | + | } | |
| 496 | + | if !scenario.cellLimited { | |
| 497 | + | return | |
| 498 | + | } | |
| 499 | + | fixture.activePane.mu.Lock() | |
| 500 | + | rows, cells, capCells := fixture.activePane.history.Len(), fixture.activePane.history.Cells(), fixture.activePane.history.CellCap() | |
| 501 | + | fixture.activePane.mu.Unlock() | |
| 502 | + | if rows >= scenario.historyRows || cells > capCells { | |
| 503 | + | b.Fatalf("293-column fixture did not enforce cell limit: rows=%d cells=%d cap=%d", rows, cells, capCells) | |
| 504 | + | } | |
| 505 | + | } | |
| 506 | + | ||
| 507 | + | func addCountingSnapshotMetrics(a, b countingSnapshotMetrics) countingSnapshotMetrics { | |
| 508 | + | return countingSnapshotMetrics{ | |
| 509 | + | writes: a.writes + b.writes, | |
| 510 | + | objectBytes: a.objectBytes + b.objectBytes, | |
| 511 | + | historyBlobBytes: a.historyBlobBytes + b.historyBlobBytes, | |
| 512 | + | manifestBytes: a.manifestBytes + b.manifestBytes, | |
| 513 | + | headBytes: a.headBytes + b.headBytes, | |
| 514 | + | } | |
| 515 | + | } | |
| 516 | + | ||
| 517 | + | func publishBenchmarkSnapshot(b testing.TB, fixture *performanceFixture, generation uint64) { | |
| 518 | + | b.Helper() | |
| 519 | + | capture, ok := fixture.d.captureSnapshotState(fixture.sess, generation) | |
| 520 | + | if !ok { | |
| 521 | + | b.Fatal("snapshot capture unexpectedly rejected benchmark fixture") | |
| 522 | + | } | |
| 523 | + | publication, err := fixture.d.incrementalPublication(capture) | |
| 524 | + | if err != nil { | |
| 525 | + | b.Fatal(err) | |
| 526 | + | } | |
| 527 | + | if err := fixture.snapshots.Publish(context.Background(), publication); err != nil { | |
| 528 | + | b.Fatal(err) | |
| 529 | + | } | |
| 530 | + | } | |
| 531 | + | ||
| 532 | + | func mutateBenchmarkVisible(fixture *performanceFixture, operation int) { | |
| 533 | + | fixture.activePane.mu.Lock() | |
| 534 | + | fixture.activePane.screen.Write([]byte(fmt.Sprintf("\x1b[1;1Hvisible-%08d", operation))) | |
| 535 | + | fixture.activePane.mu.Unlock() | |
| 536 | + | } | |
| 537 | + | ||
| 538 | + | func mutateBenchmarkTail(fixture *performanceFixture, operation int) { | |
| 539 | + | fixture.activePane.mu.Lock() | |
| 540 | + | defer fixture.activePane.mu.Unlock() | |
| 541 | + | row := make([]renderer.Cell, fixture.activePane.screen.Frame.Width) | |
| 542 | + | for column := range row { | |
| 543 | + | row[column] = renderer.Cell{Rune: rune('a' + (operation+column)%26)} | |
| 544 | + | } | |
| 545 | + | if err := fixture.activePane.history.Append(row); err != nil { | |
| 546 | + | fixture.t.Fatal(err) | |
| 547 | + | } | |
| 548 | + | } | |
| 549 | + | ||
| 550 | + | func mutateBenchmarkSealedChunk(fixture *performanceFixture, operation int) { | |
| 551 | + | for row := 0; row < 256; row++ { | |
| 552 | + | mutateBenchmarkTail(fixture, operation*256+row) | |
| 553 | + | } | |
| 554 | + | } | |
| 555 | + | ||
| 556 | + | func benchmarkSnapshotCacheBytes(sess *session) int { | |
| 557 | + | sess.snapshotMu.Lock() | |
| 558 | + | defer sess.snapshotMu.Unlock() | |
| 559 | + | if sess.snapshotChunkCache == nil { | |
| 560 | + | return 0 | |
| 561 | + | } | |
| 562 | + | return sess.snapshotChunkCache.used | |
| 563 | + | } | |
| 564 | + | ||
| 565 | + | func benchmarkReportSnapshotRepositoryMetrics(b *testing.B, metrics countingSnapshotMetrics, operations int, peakCacheBytes int) { | |
| 566 | + | b.Helper() | |
| 567 | + | if operations == 0 { | |
| 568 | + | return | |
| 390 | 569 | } | |
| 570 | + | perOperation := float64(operations) | |
| 571 | + | b.ReportMetric(float64(metrics.objectBytes)/perOperation, "objectbytes/op") | |
| 572 | + | b.ReportMetric(float64(metrics.historyBlobBytes)/perOperation, "historyblobbytes/op") | |
| 573 | + | b.ReportMetric(float64(metrics.manifestBytes)/perOperation, "manifestbytes/op") | |
| 574 | + | b.ReportMetric(float64(metrics.headBytes)/perOperation, "headbytes/op") | |
| 575 | + | b.ReportMetric(float64(peakCacheBytes), "peakcachebytes") | |
| 391 | 576 | } | |
| 392 | 577 | ||
| 393 | 578 | func benchmarkDaemonSnapshotRestore(b *testing.B) { | |
| @@ -805,7 +990,7 @@ func (f *performanceFixture) configureTab(tb *tab, tabIndex, paneCount, historyR | |||
| 805 | 990 | for _, p := range tb.panes { | |
| 806 | 991 | p.mu.Lock() | |
| 807 | 992 | width := p.screen.Frame.Width | |
| 808 | - | for row := 0; p.history.Len() < historyRows; row++ { | |
| 993 | + | for row := 0; p.history.Len() < historyRows && row < historyRows+p.screen.Frame.Height; row++ { | |
| 809 | 994 | p.screen.Write([]byte(performanceFullWidthRow(width, tabIndex, row) + "\r\n")) | |
| 810 | 995 | } | |
| 811 | 996 | p.mu.Unlock() | |
| @@ -1037,24 +1222,50 @@ func (t *countingOutputTransport) lastPayload() []byte { | |||
| 1037 | 1222 | return t.last | |
| 1038 | 1223 | } | |
| 1039 | 1224 | ||
| 1225 | + | // snapshotHeadBytes is the on-disk VEVH header (magic, generation, and | |
| 1226 | + | // manifest digest) written by the filesystem repository for every generation. | |
| 1227 | + | const snapshotHeadBytes = 4 + 8 + 32 | |
| 1228 | + | ||
| 1040 | 1229 | type countingSnapshotMetrics struct { | |
| 1041 | - | writes, manifestBytes, objectBytes uint64 | |
| 1230 | + | writes, objectBytes, historyBlobBytes, manifestBytes, headBytes uint64 | |
| 1042 | 1231 | } | |
| 1043 | 1232 | type countingSnapshotRepository struct { | |
| 1044 | 1233 | mu sync.Mutex | |
| 1045 | 1234 | countingSnapshotMetrics | |
| 1046 | - | last []byte | |
| 1235 | + | objects map[string]map[ports.SnapshotDigest]struct{} | |
| 1236 | + | last []byte | |
| 1047 | 1237 | } | |
| 1048 | 1238 | ||
| 1239 | + | // Publish models the bytes written by the filesystem repository: immutable | |
| 1240 | + | // VEVO objects are written once per session and each generation writes a VEVM | |
| 1241 | + | // manifest plus the mutable VEVH pointer. It intentionally does not measure | |
| 1242 | + | // directory metadata, temporary files, or fsyncs. | |
| 1049 | 1243 | func (s *countingSnapshotRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error { | |
| 1050 | 1244 | s.mu.Lock() | |
| 1245 | + | defer s.mu.Unlock() | |
| 1246 | + | if s.objects == nil { | |
| 1247 | + | s.objects = make(map[string]map[ports.SnapshotDigest]struct{}) | |
| 1248 | + | } | |
| 1249 | + | objects := s.objects[publication.Name] | |
| 1250 | + | if objects == nil { | |
| 1251 | + | objects = make(map[ports.SnapshotDigest]struct{}) | |
| 1252 | + | s.objects[publication.Name] = objects | |
| 1253 | + | } | |
| 1051 | 1254 | s.writes++ | |
| 1052 | 1255 | s.manifestBytes += uint64(len(publication.Manifest)) | |
| 1256 | + | s.headBytes += snapshotHeadBytes | |
| 1053 | 1257 | for _, object := range publication.Objects { | |
| 1258 | + | if _, exists := objects[object.Digest]; exists { | |
| 1259 | + | continue | |
| 1260 | + | } | |
| 1261 | + | objects[object.Digest] = struct{}{} | |
| 1054 | 1262 | s.objectBytes += uint64(len(object.Data)) | |
| 1263 | + | kind, _, err := snapcodec.PreflightObject(object.Data) | |
| 1264 | + | if err == nil && kind == snapcodec.HistoryChunk { | |
| 1265 | + | s.historyBlobBytes += uint64(len(object.Data)) | |
| 1266 | + | } | |
| 1055 | 1267 | } | |
| 1056 | 1268 | s.last = publication.Manifest | |
| 1057 | - | s.mu.Unlock() | |
| 1058 | 1269 | return nil | |
| 1059 | 1270 | } | |
| 1060 | 1271 | func (*countingSnapshotRepository) List(context.Context) ([]string, error) { return nil, nil } | |
| @@ -14,6 +14,7 @@ import ( | |||||
| 14 | "github.com/bnema/vev/internal/ports" | 14 | "github.com/bnema/vev/internal/ports" | ||
| 15 | "github.com/bnema/vev/internal/usecase/layout" | 15 | "github.com/bnema/vev/internal/usecase/layout" | ||
| 16 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | 16 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | ||
| 17 | + | "github.com/bnema/vev/pkg/renderer" | |||
| 17 | ) | 18 | ) | ||
| 18 | 19 | ||||
| 19 | func TestCountingSnapshotRepositoryPublishCountsAndRetainsManifest(t *testing.T) { | 20 | func TestCountingSnapshotRepositoryPublishCountsAndRetainsManifest(t *testing.T) { | ||
| @@ -22,13 +23,78 @@ func TestCountingSnapshotRepositoryPublishCountsAndRetainsManifest(t *testing.T) | |||||
| 22 | object := []byte{0x01, 0x02} | 23 | object := []byte{0x01, 0x02} | ||
| 23 | 24 | ||||
| 24 | require.NoError(t, repository.Publish(context.Background(), ports.SnapshotPublication{Name: "session", Manifest: payload, Objects: []ports.SnapshotObject{{Data: object}}})) | 25 | require.NoError(t, repository.Publish(context.Background(), ports.SnapshotPublication{Name: "session", Manifest: payload, Objects: []ports.SnapshotObject{{Data: object}}})) | ||
| 25 | - | require.Equal(t, countingSnapshotMetrics{writes: 1, manifestBytes: uint64(len(payload)), objectBytes: uint64(len(object))}, repository.metrics()) | 26 | + | require.Equal(t, countingSnapshotMetrics{writes: 1, manifestBytes: uint64(len(payload)), objectBytes: uint64(len(object)), headBytes: snapshotHeadBytes}, repository.metrics()) |
| 26 | require.Equal(t, payload, repository.lastPayload()) | 27 | require.Equal(t, payload, repository.lastPayload()) | ||
| 27 | 28 | ||||
| 28 | payload[0] = 0x01 | 29 | payload[0] = 0x01 | ||
| 29 | require.Equal(t, payload, repository.lastPayload(), "the counting sink must retain the payload slice header, not copy it") | 30 | require.Equal(t, payload, repository.lastPayload(), "the counting sink must retain the payload slice header, not copy it") | ||
| 30 | } | 31 | } | ||
| 31 | 32 | ||||
| 33 | + | func TestCountingSnapshotRepositoryDoesNotRewriteKnownObjects(t *testing.T) { | |||
| 34 | + | repository := &countingSnapshotRepository{} | |||
| 35 | + | object, err := snapcodec.MarshalObject(snapcodec.HistoryChunk, []byte{0x01}) | |||
| 36 | + | require.NoError(t, err) | |||
| 37 | + | publication := ports.SnapshotPublication{Name: "session", Manifest: []byte("manifest"), Objects: []ports.SnapshotObject{object}} | |||
| 38 | + | ||||
| 39 | + | require.NoError(t, repository.Publish(context.Background(), publication)) | |||
| 40 | + | repository.reset() | |||
| 41 | + | require.NoError(t, repository.Publish(context.Background(), publication)) | |||
| 42 | + | ||||
| 43 | + | require.Zero(t, repository.metrics().objectBytes, "filesystem repositories retain immutable content-addressed objects") | |||
| 44 | + | } | |||
| 45 | + | ||||
| 46 | + | func TestIncrementalSnapshotMetricsWriteNoUnchangedHistoryBlobsAndBoundCache(t *testing.T) { | |||
| 47 | + | fixture := newPerformanceFixture(t, performanceConfig{size: domain.Size{Cols: 120, Rows: 40}, panes: 1, historyRows: 10_000}) | |||
| 48 | + | ||||
| 49 | + | publishBenchmarkSnapshot(t, fixture, 1) | |||
| 50 | + | fixture.snapshots.reset() | |||
| 51 | + | publishBenchmarkSnapshot(t, fixture, 2) | |||
| 52 | + | ||||
| 53 | + | metrics := fixture.snapshots.metrics() | |||
| 54 | + | require.Zero(t, metrics.historyBlobBytes) | |||
| 55 | + | require.Zero(t, metrics.objectBytes) | |||
| 56 | + | require.Positive(t, metrics.manifestBytes) | |||
| 57 | + | require.Equal(t, uint64(snapshotHeadBytes), metrics.headBytes) | |||
| 58 | + | require.LessOrEqual(t, benchmarkSnapshotCacheBytes(fixture.sess), snapshotChunkCacheLimit) | |||
| 59 | + | } | |||
| 60 | + | ||||
| 61 | + | func TestIncrementalSnapshotMetricScenarios(t *testing.T) { | |||
| 62 | + | for _, tt := range []struct { | |||
| 63 | + | name string | |||
| 64 | + | historyRows int | |||
| 65 | + | mutate func(*performanceFixture, int) | |||
| 66 | + | wantHistory bool | |||
| 67 | + | }{ | |||
| 68 | + | {name: "tail-only", historyRows: 9_999, mutate: mutateBenchmarkTail}, | |||
| 69 | + | {name: "new-sealed-chunk", historyRows: 10_000, mutate: mutateBenchmarkSealedChunk, wantHistory: true}, | |||
| 70 | + | } { | |||
| 71 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 72 | + | fixture := newPerformanceFixture(t, performanceConfig{size: domain.Size{Cols: 120, Rows: 40}, panes: 1, historyRows: tt.historyRows}) | |||
| 73 | + | publishBenchmarkSnapshot(t, fixture, 1) | |||
| 74 | + | fixture.snapshots.reset() | |||
| 75 | + | tt.mutate(fixture, 0) | |||
| 76 | + | publishBenchmarkSnapshot(t, fixture, 2) | |||
| 77 | + | ||||
| 78 | + | metrics := fixture.snapshots.metrics() | |||
| 79 | + | if tt.wantHistory { | |||
| 80 | + | require.Positive(t, metrics.historyBlobBytes) | |||
| 81 | + | } else { | |||
| 82 | + | require.Zero(t, metrics.historyBlobBytes) | |||
| 83 | + | } | |||
| 84 | + | require.LessOrEqual(t, benchmarkSnapshotCacheBytes(fixture.sess), snapshotChunkCacheLimit) | |||
| 85 | + | }) | |||
| 86 | + | } | |||
| 87 | + | } | |||
| 88 | + | ||||
| 89 | + | func TestPerformanceFixtureCellLimitedHistory(t *testing.T) { | |||
| 90 | + | fixture := newPerformanceFixture(t, performanceConfig{size: domain.Size{Cols: 293, Rows: 40}, panes: 1, historyRows: 10_000}) | |||
| 91 | + | fixture.activePane.mu.Lock() | |||
| 92 | + | rows, cells, capCells := fixture.activePane.history.Len(), fixture.activePane.history.Cells(), fixture.activePane.history.CellCap() | |||
| 93 | + | fixture.activePane.mu.Unlock() | |||
| 94 | + | require.Less(t, rows, 10_000) | |||
| 95 | + | require.LessOrEqual(t, cells, capCells) | |||
| 96 | + | } | |||
| 97 | + | ||||
| 32 | func TestCountingOutputTransportCountsOpaquePayloadAndRejectsShortPayload(t *testing.T) { | 98 | func TestCountingOutputTransportCountsOpaquePayloadAndRejectsShortPayload(t *testing.T) { | ||
| 33 | transport := &countingOutputTransport{} | 99 | transport := &countingOutputTransport{} | ||
| 34 | 100 | ||||
| @@ -219,10 +285,16 @@ func BenchmarkDaemonHistorySnapshotCapture(b *testing.B) { | |||||
| 219 | benchmarkDaemonSnapshotCapture(b) | 285 | benchmarkDaemonSnapshotCapture(b) | ||
| 220 | } | 286 | } | ||
| 221 | 287 | ||||
| 222 | - | // BenchmarkDaemonHistorySnapshotEncode measures incremental publication from an | 288 | + | // BenchmarkDaemonIncrementalSnapshotRepository measures the complete |
| 223 | - | // already immutable capture, without queueing or persistence. | 289 | + | // content-addressed publication path against a filesystem-equivalent byte |
| 224 | - | func BenchmarkDaemonHistorySnapshotEncode(b *testing.B) { | 290 | + | // counter. Each scenario reports persisted VEVO, VEVM, and VEVH bytes, plus |
| 225 | - | benchmarkDaemonSnapshotEncode(b) | 291 | + | // allocation/time and the per-session encoded-history cache high-water mark. |
| 292 | + | func BenchmarkDaemonIncrementalSnapshotRepository(b *testing.B) { | |||
| 293 | + | for _, scenario := range snapshotBenchmarkScenarios { | |||
| 294 | + | b.Run(scenario.name, func(b *testing.B) { | |||
| 295 | + | benchmarkIncrementalSnapshotRepository(b, scenario) | |||
| 296 | + | }) | |||
| 297 | + | } | |||
| 226 | } | 298 | } | ||
| 227 | 299 | ||||
| 228 | // BenchmarkDaemonHistorySnapshotRestore measures terminal-state restoration | 300 | // BenchmarkDaemonHistorySnapshotRestore measures terminal-state restoration | ||
| @@ -351,43 +423,156 @@ func benchmarkDaemonSnapshotCapture(b *testing.B) { | |||||
| 351 | } | 423 | } | ||
| 352 | } | 424 | } | ||
| 353 | 425 | ||||
| 354 | - | func benchmarkDaemonSnapshotEncode(b *testing.B) { | 426 | + | type snapshotBenchmarkScenario struct { |
| 355 | - | for _, topology := range daemonSnapshotTopologies { | 427 | + | name string |
| 356 | - | b.Run(topology.name, func(b *testing.B) { | 428 | + | size domain.Size |
| 357 | - | fixture := newPerformanceFixture(b, performanceConfig{size: domain.Size{Cols: 120, Rows: 40}, tabs: topology.tabs, panes: topology.panes, historyRows: 10_000}) | 429 | + | historyRows int |
| 358 | - | if !fixture.hasHistoryTopology(topology.tabs, topology.panes, 10_000) { | 430 | + | baseline bool |
| 359 | - | b.Fatal("invalid daemon snapshot publication fixture") | 431 | + | mutate func(*performanceFixture, int) |
| 360 | - | } | 432 | + | unchanged bool |
| 361 | - | capture, ok := fixture.d.captureSnapshotState(fixture.sess, 1) | 433 | + | cellLimited bool |
| 362 | - | if !ok || !validBenchmarkCapture(capture) { | 434 | + | } |
| 363 | - | b.Fatal("invalid snapshot publication fixture") | 435 | + | |
| 364 | - | } | 436 | + | var snapshotBenchmarkScenarios = []snapshotBenchmarkScenario{ |
| 365 | - | publication, err := fixture.d.incrementalPublication(capture) | 437 | + | {name: "initial-10k-x-120", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 10_000}, |
| 366 | - | if err != nil { | 438 | + | {name: "unchanged", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 10_000, baseline: true, unchanged: true}, |
| 367 | - | b.Fatal(err) | 439 | + | {name: "visible-only", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 10_000, baseline: true, mutate: mutateBenchmarkVisible}, |
| 368 | - | } | 440 | + | // Leave one row below the retention cap so the mutation changes only the |
| 369 | - | if _, err := snapcodec.UnmarshalManifest(publication.Manifest); err != nil || len(publication.Objects) == 0 { | 441 | + | // mutable tail and cannot evict a sealed chunk. |
| 370 | - | b.Fatalf("invalid incremental snapshot publication fixture: %v", err) | 442 | + | {name: "tail-only", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 9_999, baseline: true, mutate: mutateBenchmarkTail}, |
| 371 | - | } | 443 | + | {name: "new-sealed-chunk", size: domain.Size{Cols: 120, Rows: 40}, historyRows: 10_000, baseline: true, mutate: mutateBenchmarkSealedChunk}, |
| 372 | - | var last ports.SnapshotPublication | 444 | + | {name: "cell-limited-293-columns", size: domain.Size{Cols: 293, Rows: 40}, historyRows: 10_000, cellLimited: true}, |
| 373 | - | b.ReportAllocs() | 445 | + | } |
| 374 | - | b.SetBytes(int64(snapshotPublicationBytes(publication))) | 446 | + | |
| 375 | - | b.ResetTimer() | 447 | + | func benchmarkIncrementalSnapshotRepository(b *testing.B, scenario snapshotBenchmarkScenario) { |
| 376 | - | for range b.N { | 448 | + | var ( |
| 377 | - | last, err = fixture.d.incrementalPublication(capture) | 449 | + | metrics countingSnapshotMetrics |
| 378 | - | if err != nil { | 450 | + | peakCacheBytes int |
| 379 | - | b.Fatal(err) | 451 | + | ) |
| 380 | - | } | 452 | + | b.ReportAllocs() |
| 381 | - | } | 453 | + | b.ResetTimer() |
| 382 | - | b.StopTimer() | 454 | + | for operation := 0; b.Loop(); operation++ { |
| 383 | - | if _, err := snapcodec.UnmarshalManifest(last.Manifest); err != nil { | 455 | + | // A new fixture is deliberately outside the timed region. It gives every |
| 384 | - | b.Fatalf("snapshot publication produced invalid manifest: %v", err) | 456 | + | // iteration the same filesystem state: initial has no generation, |
| 385 | - | } | 457 | + | // unchanged/visible/tail/sealed begin after one persisted generation. |
| 386 | - | b.ReportMetric(float64(len(last.Manifest)), "manifestbytes/op") | 458 | + | // Reusing one fixture would eventually turn a tail into a sealed chunk |
| 387 | - | b.ReportMetric(float64(snapshotObjectBytes(last)), "objectbytes/op") | 459 | + | // and dilute initial-write bytes over later idempotent publications. |
| 388 | - | b.ReportMetric(float64(len(last.Objects)), "objects/op") | 460 | + | b.StopTimer() |
| 389 | - | }) | 461 | + | fixture := newPerformanceFixture(b, performanceConfig{size: scenario.size, panes: 1, historyRows: scenario.historyRows}) |
| 462 | + | validateSnapshotBenchmarkFixture(b, fixture, scenario) | |||
| 463 | + | generation := uint64(1) | |||
| 464 | + | if scenario.baseline { | |||
| 465 | + | publishBenchmarkSnapshot(b, fixture, generation) | |||
| 466 | + | generation++ | |||
| 467 | + | fixture.snapshots.reset() | |||
| 468 | + | } | |||
| 469 | + | b.StartTimer() | |||
| 470 | + | if scenario.mutate != nil { | |||
| 471 | + | scenario.mutate(fixture, operation) | |||
| 472 | + | } | |||
| 473 | + | publishBenchmarkSnapshot(b, fixture, generation) | |||
| 474 | + | b.StopTimer() | |||
| 475 | + | ||||
| 476 | + | metrics = addCountingSnapshotMetrics(metrics, fixture.snapshots.metrics()) | |||
| 477 | + | peakCacheBytes = max(peakCacheBytes, benchmarkSnapshotCacheBytes(fixture.sess)) | |||
| 478 | + | // B.Loop requires the timer to be running at the next iteration. | |||
| 479 | + | b.StartTimer() | |||
| 480 | + | } | |||
| 481 | + | b.StopTimer() | |||
| 482 | + | if scenario.unchanged && metrics.historyBlobBytes != 0 { | |||
| 483 | + | b.Fatalf("unchanged snapshot wrote %d history blob bytes", metrics.historyBlobBytes) | |||
| 484 | + | } | |||
| 485 | + | if peakCacheBytes > snapshotChunkCacheLimit { | |||
| 486 | + | b.Fatalf("snapshot cache peak = %d, limit = %d", peakCacheBytes, snapshotChunkCacheLimit) | |||
| 487 | + | } | |||
| 488 | + | benchmarkReportSnapshotRepositoryMetrics(b, metrics, b.N, peakCacheBytes) | |||
| 489 | + | } | |||
| 490 | + | ||||
| 491 | + | func validateSnapshotBenchmarkFixture(b *testing.B, fixture *performanceFixture, scenario snapshotBenchmarkScenario) { | |||
| 492 | + | b.Helper() | |||
| 493 | + | if !fixture.hasHistoryTopology(1, 1, scenario.historyRows) && !scenario.cellLimited { | |||
| 494 | + | b.Fatal("invalid incremental snapshot fixture") | |||
| 495 | + | } | |||
| 496 | + | if !scenario.cellLimited { | |||
| 497 | + | return | |||
| 498 | + | } | |||
| 499 | + | fixture.activePane.mu.Lock() | |||
| 500 | + | rows, cells, capCells := fixture.activePane.history.Len(), fixture.activePane.history.Cells(), fixture.activePane.history.CellCap() | |||
| 501 | + | fixture.activePane.mu.Unlock() | |||
| 502 | + | if rows >= scenario.historyRows || cells > capCells { | |||
| 503 | + | b.Fatalf("293-column fixture did not enforce cell limit: rows=%d cells=%d cap=%d", rows, cells, capCells) | |||
| 504 | + | } | |||
| 505 | + | } | |||
| 506 | + | ||||
| 507 | + | func addCountingSnapshotMetrics(a, b countingSnapshotMetrics) countingSnapshotMetrics { | |||
| 508 | + | return countingSnapshotMetrics{ | |||
| 509 | + | writes: a.writes + b.writes, | |||
| 510 | + | objectBytes: a.objectBytes + b.objectBytes, | |||
| 511 | + | historyBlobBytes: a.historyBlobBytes + b.historyBlobBytes, | |||
| 512 | + | manifestBytes: a.manifestBytes + b.manifestBytes, | |||
| 513 | + | headBytes: a.headBytes + b.headBytes, | |||
| 514 | + | } | |||
| 515 | + | } | |||
| 516 | + | ||||
| 517 | + | func publishBenchmarkSnapshot(b testing.TB, fixture *performanceFixture, generation uint64) { | |||
| 518 | + | b.Helper() | |||
| 519 | + | capture, ok := fixture.d.captureSnapshotState(fixture.sess, generation) | |||
| 520 | + | if !ok { | |||
| 521 | + | b.Fatal("snapshot capture unexpectedly rejected benchmark fixture") | |||
| 522 | + | } | |||
| 523 | + | publication, err := fixture.d.incrementalPublication(capture) | |||
| 524 | + | if err != nil { | |||
| 525 | + | b.Fatal(err) | |||
| 526 | + | } | |||
| 527 | + | if err := fixture.snapshots.Publish(context.Background(), publication); err != nil { | |||
| 528 | + | b.Fatal(err) | |||
| 529 | + | } | |||
| 530 | + | } | |||
| 531 | + | ||||
| 532 | + | func mutateBenchmarkVisible(fixture *performanceFixture, operation int) { | |||
| 533 | + | fixture.activePane.mu.Lock() | |||
| 534 | + | fixture.activePane.screen.Write([]byte(fmt.Sprintf("\x1b[1;1Hvisible-%08d", operation))) | |||
| 535 | + | fixture.activePane.mu.Unlock() | |||
| 536 | + | } | |||
| 537 | + | ||||
| 538 | + | func mutateBenchmarkTail(fixture *performanceFixture, operation int) { | |||
| 539 | + | fixture.activePane.mu.Lock() | |||
| 540 | + | defer fixture.activePane.mu.Unlock() | |||
| 541 | + | row := make([]renderer.Cell, fixture.activePane.screen.Frame.Width) | |||
| 542 | + | for column := range row { | |||
| 543 | + | row[column] = renderer.Cell{Rune: rune('a' + (operation+column)%26)} | |||
| 544 | + | } | |||
| 545 | + | if err := fixture.activePane.history.Append(row); err != nil { | |||
| 546 | + | fixture.t.Fatal(err) | |||
| 547 | + | } | |||
| 548 | + | } | |||
| 549 | + | ||||
| 550 | + | func mutateBenchmarkSealedChunk(fixture *performanceFixture, operation int) { | |||
| 551 | + | for row := 0; row < 256; row++ { | |||
| 552 | + | mutateBenchmarkTail(fixture, operation*256+row) | |||
| 553 | + | } | |||
| 554 | + | } | |||
| 555 | + | ||||
| 556 | + | func benchmarkSnapshotCacheBytes(sess *session) int { | |||
| 557 | + | sess.snapshotMu.Lock() | |||
| 558 | + | defer sess.snapshotMu.Unlock() | |||
| 559 | + | if sess.snapshotChunkCache == nil { | |||
| 560 | + | return 0 | |||
| 561 | + | } | |||
| 562 | + | return sess.snapshotChunkCache.used | |||
| 563 | + | } | |||
| 564 | + | ||||
| 565 | + | func benchmarkReportSnapshotRepositoryMetrics(b *testing.B, metrics countingSnapshotMetrics, operations int, peakCacheBytes int) { | |||
| 566 | + | b.Helper() | |||
| 567 | + | if operations == 0 { | |||
| 568 | + | return | |||
| 390 | } | 569 | } | ||
| 570 | + | perOperation := float64(operations) | |||
| 571 | + | b.ReportMetric(float64(metrics.objectBytes)/perOperation, "objectbytes/op") | |||
| 572 | + | b.ReportMetric(float64(metrics.historyBlobBytes)/perOperation, "historyblobbytes/op") | |||
| 573 | + | b.ReportMetric(float64(metrics.manifestBytes)/perOperation, "manifestbytes/op") | |||
| 574 | + | b.ReportMetric(float64(metrics.headBytes)/perOperation, "headbytes/op") | |||
| 575 | + | b.ReportMetric(float64(peakCacheBytes), "peakcachebytes") | |||
| 391 | } | 576 | } | ||
| 392 | 577 | ||||
| 393 | func benchmarkDaemonSnapshotRestore(b *testing.B) { | 578 | func benchmarkDaemonSnapshotRestore(b *testing.B) { | ||
| @@ -805,7 +990,7 @@ func (f *performanceFixture) configureTab(tb *tab, tabIndex, paneCount, historyR | |||||
| 805 | for _, p := range tb.panes { | 990 | for _, p := range tb.panes { | ||
| 806 | p.mu.Lock() | 991 | p.mu.Lock() | ||
| 807 | width := p.screen.Frame.Width | 992 | width := p.screen.Frame.Width | ||
| 808 | - | for row := 0; p.history.Len() < historyRows; row++ { | 993 | + | for row := 0; p.history.Len() < historyRows && row < historyRows+p.screen.Frame.Height; row++ { |
| 809 | p.screen.Write([]byte(performanceFullWidthRow(width, tabIndex, row) + "\r\n")) | 994 | p.screen.Write([]byte(performanceFullWidthRow(width, tabIndex, row) + "\r\n")) | ||
| 810 | } | 995 | } | ||
| 811 | p.mu.Unlock() | 996 | p.mu.Unlock() | ||
| @@ -1037,24 +1222,50 @@ func (t *countingOutputTransport) lastPayload() []byte { | |||||
| 1037 | return t.last | 1222 | return t.last | ||
| 1038 | } | 1223 | } | ||
| 1039 | 1224 | ||||
| 1225 | + | // snapshotHeadBytes is the on-disk VEVH header (magic, generation, and | |||
| 1226 | + | // manifest digest) written by the filesystem repository for every generation. | |||
| 1227 | + | const snapshotHeadBytes = 4 + 8 + 32 | |||
| 1228 | + | ||||
| 1040 | type countingSnapshotMetrics struct { | 1229 | type countingSnapshotMetrics struct { | ||
| 1041 | - | writes, manifestBytes, objectBytes uint64 | 1230 | + | writes, objectBytes, historyBlobBytes, manifestBytes, headBytes uint64 |
| 1042 | } | 1231 | } | ||
| 1043 | type countingSnapshotRepository struct { | 1232 | type countingSnapshotRepository struct { | ||
| 1044 | mu sync.Mutex | 1233 | mu sync.Mutex | ||
| 1045 | countingSnapshotMetrics | 1234 | countingSnapshotMetrics | ||
| 1046 | - | last []byte | 1235 | + | objects map[string]map[ports.SnapshotDigest]struct{} |
| 1236 | + | last []byte | |||
| 1047 | } | 1237 | } | ||
| 1048 | 1238 | ||||
| 1239 | + | // Publish models the bytes written by the filesystem repository: immutable | |||
| 1240 | + | // VEVO objects are written once per session and each generation writes a VEVM | |||
| 1241 | + | // manifest plus the mutable VEVH pointer. It intentionally does not measure | |||
| 1242 | + | // directory metadata, temporary files, or fsyncs. | |||
| 1049 | func (s *countingSnapshotRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error { | 1243 | func (s *countingSnapshotRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error { | ||
| 1050 | s.mu.Lock() | 1244 | s.mu.Lock() | ||
| 1245 | + | defer s.mu.Unlock() | |||
| 1246 | + | if s.objects == nil { | |||
| 1247 | + | s.objects = make(map[string]map[ports.SnapshotDigest]struct{}) | |||
| 1248 | + | } | |||
| 1249 | + | objects := s.objects[publication.Name] | |||
| 1250 | + | if objects == nil { | |||
| 1251 | + | objects = make(map[ports.SnapshotDigest]struct{}) | |||
| 1252 | + | s.objects[publication.Name] = objects | |||
| 1253 | + | } | |||
| 1051 | s.writes++ | 1254 | s.writes++ | ||
| 1052 | s.manifestBytes += uint64(len(publication.Manifest)) | 1255 | s.manifestBytes += uint64(len(publication.Manifest)) | ||
| 1256 | + | s.headBytes += snapshotHeadBytes | |||
| 1053 | for _, object := range publication.Objects { | 1257 | for _, object := range publication.Objects { | ||
| 1258 | + | if _, exists := objects[object.Digest]; exists { | |||
| 1259 | + | continue | |||
| 1260 | + | } | |||
| 1261 | + | objects[object.Digest] = struct{}{} | |||
| 1054 | s.objectBytes += uint64(len(object.Data)) | 1262 | s.objectBytes += uint64(len(object.Data)) | ||
| 1263 | + | kind, _, err := snapcodec.PreflightObject(object.Data) | |||
| 1264 | + | if err == nil && kind == snapcodec.HistoryChunk { | |||
| 1265 | + | s.historyBlobBytes += uint64(len(object.Data)) | |||
| 1266 | + | } | |||
| 1055 | } | 1267 | } | ||
| 1056 | s.last = publication.Manifest | 1268 | s.last = publication.Manifest | ||
| 1057 | - | s.mu.Unlock() | |||
| 1058 | return nil | 1269 | return nil | ||
| 1059 | } | 1270 | } | ||
| 1060 | func (*countingSnapshotRepository) List(context.Context) ([]string, error) { return nil, nil } | 1271 | func (*countingSnapshotRepository) List(context.Context) ([]string, error) { return nil, nil } | ||
D internal/usecase/snapshot/benchmark_test.go +0 -124
| @@ -1,124 +0,0 @@ | |||
| 1 | - | package snapshot | |
| 2 | - | ||
| 3 | - | import ( | |
| 4 | - | "testing" | |
| 5 | - | ||
| 6 | - | "github.com/bnema/vev/internal/usecase/layout" | |
| 7 | - | "github.com/bnema/vev/pkg/renderer" | |
| 8 | - | "github.com/bnema/vev/pkg/vt" | |
| 9 | - | ) | |
| 10 | - | ||
| 11 | - | var ( | |
| 12 | - | benchmarkV3BytesSink []byte | |
| 13 | - | benchmarkV3SessionSink Session | |
| 14 | - | ) | |
| 15 | - | ||
| 16 | - | func BenchmarkSnapshotV3Marshal10KRows(b *testing.B) { | |
| 17 | - | session, encoded := benchmarkV3Session(b) | |
| 18 | - | b.ReportAllocs() | |
| 19 | - | b.SetBytes(int64(len(encoded))) | |
| 20 | - | b.ResetTimer() | |
| 21 | - | for b.Loop() { | |
| 22 | - | var err error | |
| 23 | - | benchmarkV3BytesSink, err = Marshal(session) | |
| 24 | - | if err != nil { | |
| 25 | - | b.Fatal(err) | |
| 26 | - | } | |
| 27 | - | } | |
| 28 | - | b.StopTimer() | |
| 29 | - | if _, err := Unmarshal(benchmarkV3BytesSink); err != nil { | |
| 30 | - | b.Fatalf("marshal produced an invalid v3 snapshot: %v", err) | |
| 31 | - | } | |
| 32 | - | } | |
| 33 | - | ||
| 34 | - | func BenchmarkSnapshotV3Unmarshal10KRows(b *testing.B) { | |
| 35 | - | _, encoded := benchmarkV3Session(b) | |
| 36 | - | b.ReportAllocs() | |
| 37 | - | b.SetBytes(int64(len(encoded))) | |
| 38 | - | b.ResetTimer() | |
| 39 | - | for b.Loop() { | |
| 40 | - | var err error | |
| 41 | - | benchmarkV3SessionSink, err = Unmarshal(encoded) | |
| 42 | - | if err != nil { | |
| 43 | - | b.Fatal(err) | |
| 44 | - | } | |
| 45 | - | } | |
| 46 | - | b.StopTimer() | |
| 47 | - | if len(benchmarkV3SessionSink.Tabs) != 1 || len(benchmarkV3SessionSink.Tabs[0].Panes) != 1 || len(benchmarkV3SessionSink.Tabs[0].Panes[0].SealedChunks) != 40 { | |
| 48 | - | b.Fatal("unmarshal lost canonical 10k-row snapshot data") | |
| 49 | - | } | |
| 50 | - | } | |
| 51 | - | ||
| 52 | - | func benchmarkV3Session(t testing.TB) (Session, []byte) { | |
| 53 | - | t.Helper() | |
| 54 | - | const ( | |
| 55 | - | cols = 120 | |
| 56 | - | rows = 40 | |
| 57 | - | historyRows = 10_000 | |
| 58 | - | chunkRows = 256 | |
| 59 | - | ) | |
| 60 | - | history := vt.NewHistory(vt.HistoryConfig{MaxRows: historyRows, ChunkRows: chunkRows}) | |
| 61 | - | for row := range historyRows { | |
| 62 | - | cells := make([]renderer.Cell, cols) | |
| 63 | - | for col := range cells { | |
| 64 | - | cells[col] = renderer.Cell{Rune: rune('a' + (row+col)%26)} | |
| 65 | - | } | |
| 66 | - | if err := history.Append(cells); err != nil { | |
| 67 | - | t.Fatal(err) | |
| 68 | - | } | |
| 69 | - | } | |
| 70 | - | sealed, tail, err := vt.MarshalSealedHistory(history.SealAndView()) | |
| 71 | - | if err != nil { | |
| 72 | - | t.Fatal(err) | |
| 73 | - | } | |
| 74 | - | visibleFrame := renderer.NewFrame(cols, rows) | |
| 75 | - | for y := range rows { | |
| 76 | - | for x := range cols { | |
| 77 | - | visibleFrame.Set(x, y, renderer.Cell{Rune: rune('A' + (x+y)%26)}) | |
| 78 | - | } | |
| 79 | - | } | |
| 80 | - | visible, err := vt.MarshalVisible(visibleFrame) | |
| 81 | - | if err != nil { | |
| 82 | - | t.Fatal(err) | |
| 83 | - | } | |
| 84 | - | session := Session{ | |
| 85 | - | Name: "benchmark", | |
| 86 | - | CreatedAt: 1, | |
| 87 | - | Tabs: []Tab{{ | |
| 88 | - | StableID: "tab-1", | |
| 89 | - | Cols: cols, | |
| 90 | - | Rows: rows, | |
| 91 | - | NextPaneID: 2, | |
| 92 | - | Focus: "pane-1", | |
| 93 | - | Tree: layout.NewTree("pane-1"), | |
| 94 | - | Panes: []Pane{{ | |
| 95 | - | ID: "pane-1", | |
| 96 | - | StableID: "pane-1", | |
| 97 | - | Cwd: "/benchmark", | |
| 98 | - | SealedChunks: sealed, | |
| 99 | - | Tail: tail, | |
| 100 | - | Visible: visible, | |
| 101 | - | }}, | |
| 102 | - | }}, | |
| 103 | - | } | |
| 104 | - | encoded, err := Marshal(session) | |
| 105 | - | if err != nil { | |
| 106 | - | t.Fatal(err) | |
| 107 | - | } | |
| 108 | - | decoded, err := Unmarshal(encoded) | |
| 109 | - | if err != nil { | |
| 110 | - | t.Fatal(err) | |
| 111 | - | } | |
| 112 | - | paneCount := 0 | |
| 113 | - | decodedChunks := 0 | |
| 114 | - | if len(decoded.Tabs) == 1 { | |
| 115 | - | paneCount = len(decoded.Tabs[0].Panes) | |
| 116 | - | if paneCount == 1 { | |
| 117 | - | decodedChunks = len(decoded.Tabs[0].Panes[0].SealedChunks) | |
| 118 | - | } | |
| 119 | - | } | |
| 120 | - | if len(sealed) != (historyRows+chunkRows-1)/chunkRows || paneCount != 1 || decodedChunks != len(sealed) { | |
| 121 | - | t.Fatalf("invalid v3 benchmark fixture: chunks=%d tabs=%d panes=%d decoded chunks=%d", len(sealed), len(decoded.Tabs), paneCount, decodedChunks) | |
| 122 | - | } | |
| 123 | - | return session, encoded | |
| 124 | - | } | |
| @@ -1,124 +0,0 @@ | |||||
| 1 | - | package snapshot | |||
| 2 | - | ||||
| 3 | - | import ( | |||
| 4 | - | "testing" | |||
| 5 | - | ||||
| 6 | - | "github.com/bnema/vev/internal/usecase/layout" | |||
| 7 | - | "github.com/bnema/vev/pkg/renderer" | |||
| 8 | - | "github.com/bnema/vev/pkg/vt" | |||
| 9 | - | ) | |||
| 10 | - | ||||
| 11 | - | var ( | |||
| 12 | - | benchmarkV3BytesSink []byte | |||
| 13 | - | benchmarkV3SessionSink Session | |||
| 14 | - | ) | |||
| 15 | - | ||||
| 16 | - | func BenchmarkSnapshotV3Marshal10KRows(b *testing.B) { | |||
| 17 | - | session, encoded := benchmarkV3Session(b) | |||
| 18 | - | b.ReportAllocs() | |||
| 19 | - | b.SetBytes(int64(len(encoded))) | |||
| 20 | - | b.ResetTimer() | |||
| 21 | - | for b.Loop() { | |||
| 22 | - | var err error | |||
| 23 | - | benchmarkV3BytesSink, err = Marshal(session) | |||
| 24 | - | if err != nil { | |||
| 25 | - | b.Fatal(err) | |||
| 26 | - | } | |||
| 27 | - | } | |||
| 28 | - | b.StopTimer() | |||
| 29 | - | if _, err := Unmarshal(benchmarkV3BytesSink); err != nil { | |||
| 30 | - | b.Fatalf("marshal produced an invalid v3 snapshot: %v", err) | |||
| 31 | - | } | |||
| 32 | - | } | |||
| 33 | - | ||||
| 34 | - | func BenchmarkSnapshotV3Unmarshal10KRows(b *testing.B) { | |||
| 35 | - | _, encoded := benchmarkV3Session(b) | |||
| 36 | - | b.ReportAllocs() | |||
| 37 | - | b.SetBytes(int64(len(encoded))) | |||
| 38 | - | b.ResetTimer() | |||
| 39 | - | for b.Loop() { | |||
| 40 | - | var err error | |||
| 41 | - | benchmarkV3SessionSink, err = Unmarshal(encoded) | |||
| 42 | - | if err != nil { | |||
| 43 | - | b.Fatal(err) | |||
| 44 | - | } | |||
| 45 | - | } | |||
| 46 | - | b.StopTimer() | |||
| 47 | - | if len(benchmarkV3SessionSink.Tabs) != 1 || len(benchmarkV3SessionSink.Tabs[0].Panes) != 1 || len(benchmarkV3SessionSink.Tabs[0].Panes[0].SealedChunks) != 40 { | |||
| 48 | - | b.Fatal("unmarshal lost canonical 10k-row snapshot data") | |||
| 49 | - | } | |||
| 50 | - | } | |||
| 51 | - | ||||
| 52 | - | func benchmarkV3Session(t testing.TB) (Session, []byte) { | |||
| 53 | - | t.Helper() | |||
| 54 | - | const ( | |||
| 55 | - | cols = 120 | |||
| 56 | - | rows = 40 | |||
| 57 | - | historyRows = 10_000 | |||
| 58 | - | chunkRows = 256 | |||
| 59 | - | ) | |||
| 60 | - | history := vt.NewHistory(vt.HistoryConfig{MaxRows: historyRows, ChunkRows: chunkRows}) | |||
| 61 | - | for row := range historyRows { | |||
| 62 | - | cells := make([]renderer.Cell, cols) | |||
| 63 | - | for col := range cells { | |||
| 64 | - | cells[col] = renderer.Cell{Rune: rune('a' + (row+col)%26)} | |||
| 65 | - | } | |||
| 66 | - | if err := history.Append(cells); err != nil { | |||
| 67 | - | t.Fatal(err) | |||
| 68 | - | } | |||
| 69 | - | } | |||
| 70 | - | sealed, tail, err := vt.MarshalSealedHistory(history.SealAndView()) | |||
| 71 | - | if err != nil { | |||
| 72 | - | t.Fatal(err) | |||
| 73 | - | } | |||
| 74 | - | visibleFrame := renderer.NewFrame(cols, rows) | |||
| 75 | - | for y := range rows { | |||
| 76 | - | for x := range cols { | |||
| 77 | - | visibleFrame.Set(x, y, renderer.Cell{Rune: rune('A' + (x+y)%26)}) | |||
| 78 | - | } | |||
| 79 | - | } | |||
| 80 | - | visible, err := vt.MarshalVisible(visibleFrame) | |||
| 81 | - | if err != nil { | |||
| 82 | - | t.Fatal(err) | |||
| 83 | - | } | |||
| 84 | - | session := Session{ | |||
| 85 | - | Name: "benchmark", | |||
| 86 | - | CreatedAt: 1, | |||
| 87 | - | Tabs: []Tab{{ | |||
| 88 | - | StableID: "tab-1", | |||
| 89 | - | Cols: cols, | |||
| 90 | - | Rows: rows, | |||
| 91 | - | NextPaneID: 2, | |||
| 92 | - | Focus: "pane-1", | |||
| 93 | - | Tree: layout.NewTree("pane-1"), | |||
| 94 | - | Panes: []Pane{{ | |||
| 95 | - | ID: "pane-1", | |||
| 96 | - | StableID: "pane-1", | |||
| 97 | - | Cwd: "/benchmark", | |||
| 98 | - | SealedChunks: sealed, | |||
| 99 | - | Tail: tail, | |||
| 100 | - | Visible: visible, | |||
| 101 | - | }}, | |||
| 102 | - | }}, | |||
| 103 | - | } | |||
| 104 | - | encoded, err := Marshal(session) | |||
| 105 | - | if err != nil { | |||
| 106 | - | t.Fatal(err) | |||
| 107 | - | } | |||
| 108 | - | decoded, err := Unmarshal(encoded) | |||
| 109 | - | if err != nil { | |||
| 110 | - | t.Fatal(err) | |||
| 111 | - | } | |||
| 112 | - | paneCount := 0 | |||
| 113 | - | decodedChunks := 0 | |||
| 114 | - | if len(decoded.Tabs) == 1 { | |||
| 115 | - | paneCount = len(decoded.Tabs[0].Panes) | |||
| 116 | - | if paneCount == 1 { | |||
| 117 | - | decodedChunks = len(decoded.Tabs[0].Panes[0].SealedChunks) | |||
| 118 | - | } | |||
| 119 | - | } | |||
| 120 | - | if len(sealed) != (historyRows+chunkRows-1)/chunkRows || paneCount != 1 || decodedChunks != len(sealed) { | |||
| 121 | - | t.Fatalf("invalid v3 benchmark fixture: chunks=%d tabs=%d panes=%d decoded chunks=%d", len(sealed), len(decoded.Tabs), paneCount, decodedChunks) | |||
| 122 | - | } | |||
| 123 | - | return session, encoded | |||
| 124 | - | } | |||