test: add transport replay pipeline coverage
- commit
- f68fdb776501c64882044786037fee39a4f65fc0
- author
- brice · · 2026-07-12 05:23 UTC
- parent
- 72230e6
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
6 files · +270 -8
A internal/adapters/dgram/transport_replay_test.go +41 -0
| @@ -0,0 +1,41 @@ | |||
| 1 | + | package dgram | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "testing" | |
| 5 | + | ||
| 6 | + | "github.com/bnema/vev/internal/ports" | |
| 7 | + | "github.com/stretchr/testify/require" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | // This must stay byte-for-byte equal to the carriage transcript in the stream | |
| 11 | + | // transport adapters. Datagram framing is not a rendering operation. | |
| 12 | + | func replayTranscript() []ports.Frame { | |
| 13 | + | return []ports.Frame{ | |
| 14 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\x1b[2J\x1b[Hone\r\ntwo")})}, | |
| 15 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\x1b[2;1HTWO")})}, | |
| 16 | + | } | |
| 17 | + | } | |
| 18 | + | ||
| 19 | + | func TestTransportReplay(t *testing.T) { | |
| 20 | + | link := newSimulatedLink(fixedClock{}, packetPolicy{}) | |
| 21 | + | leftPC, rightPC := newPairWithCapacity(link, 32) | |
| 22 | + | left, err := NewTransport(leftPC, testAddr("b"), key(), 1, 2) | |
| 23 | + | require.NoError(t, err) | |
| 24 | + | right, err := NewTransport(rightPC, testAddr("a"), key(), 2, 1) | |
| 25 | + | require.NoError(t, err) | |
| 26 | + | t.Cleanup(func() { _ = left.Close(); _ = right.Close() }) | |
| 27 | + | ||
| 28 | + | for _, want := range replayTranscript() { | |
| 29 | + | require.NoError(t, left.Send(want)) | |
| 30 | + | link.flush(rightPC) | |
| 31 | + | got, err := right.Recv() | |
| 32 | + | require.NoError(t, err) | |
| 33 | + | require.Equal(t, want.Type, got.Type) | |
| 34 | + | require.Equal(t, want.Payload, got.Payload) | |
| 35 | + | out, err := ports.UnmarshalOutput(got.Payload) | |
| 36 | + | require.NoError(t, err) | |
| 37 | + | wantOut, err := ports.UnmarshalOutput(want.Payload) | |
| 38 | + | require.NoError(t, err) | |
| 39 | + | require.Equal(t, wantOut, out) | |
| 40 | + | } | |
| 41 | + | } | |
| @@ -0,0 +1,41 @@ | |||||
| 1 | + | package dgram | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "testing" | |||
| 5 | + | ||||
| 6 | + | "github.com/bnema/vev/internal/ports" | |||
| 7 | + | "github.com/stretchr/testify/require" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | // This must stay byte-for-byte equal to the carriage transcript in the stream | |||
| 11 | + | // transport adapters. Datagram framing is not a rendering operation. | |||
| 12 | + | func replayTranscript() []ports.Frame { | |||
| 13 | + | return []ports.Frame{ | |||
| 14 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\x1b[2J\x1b[Hone\r\ntwo")})}, | |||
| 15 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\x1b[2;1HTWO")})}, | |||
| 16 | + | } | |||
| 17 | + | } | |||
| 18 | + | ||||
| 19 | + | func TestTransportReplay(t *testing.T) { | |||
| 20 | + | link := newSimulatedLink(fixedClock{}, packetPolicy{}) | |||
| 21 | + | leftPC, rightPC := newPairWithCapacity(link, 32) | |||
| 22 | + | left, err := NewTransport(leftPC, testAddr("b"), key(), 1, 2) | |||
| 23 | + | require.NoError(t, err) | |||
| 24 | + | right, err := NewTransport(rightPC, testAddr("a"), key(), 2, 1) | |||
| 25 | + | require.NoError(t, err) | |||
| 26 | + | t.Cleanup(func() { _ = left.Close(); _ = right.Close() }) | |||
| 27 | + | ||||
| 28 | + | for _, want := range replayTranscript() { | |||
| 29 | + | require.NoError(t, left.Send(want)) | |||
| 30 | + | link.flush(rightPC) | |||
| 31 | + | got, err := right.Recv() | |||
| 32 | + | require.NoError(t, err) | |||
| 33 | + | require.Equal(t, want.Type, got.Type) | |||
| 34 | + | require.Equal(t, want.Payload, got.Payload) | |||
| 35 | + | out, err := ports.UnmarshalOutput(got.Payload) | |||
| 36 | + | require.NoError(t, err) | |||
| 37 | + | wantOut, err := ports.UnmarshalOutput(want.Payload) | |||
| 38 | + | require.NoError(t, err) | |||
| 39 | + | require.Equal(t, wantOut, out) | |||
| 40 | + | } | |||
| 41 | + | } | |||
A internal/adapters/ipc/transport_replay_test.go +48 -0
| @@ -0,0 +1,48 @@ | |||
| 1 | + | package ipc | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "net" | |
| 5 | + | "sync" | |
| 6 | + | "testing" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/vev/internal/ports" | |
| 9 | + | "github.com/stretchr/testify/require" | |
| 10 | + | ) | |
| 11 | + | ||
| 12 | + | // replayTranscript is deliberately transport-agnostic: adapters only carry | |
| 13 | + | // these already-composed output frames and must not interpret them. | |
| 14 | + | func replayTranscript() []ports.Frame { | |
| 15 | + | return []ports.Frame{ | |
| 16 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\x1b[2J\x1b[Hone\r\ntwo")})}, | |
| 17 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\x1b[2;1HTWO")})}, | |
| 18 | + | } | |
| 19 | + | } | |
| 20 | + | ||
| 21 | + | func TestTransportReplay(t *testing.T) { | |
| 22 | + | left, right := net.Pipe() | |
| 23 | + | t.Cleanup(func() { _ = left.Close(); _ = right.Close() }) | |
| 24 | + | sender, receiver := NewTransport(left), NewTransport(right) | |
| 25 | + | frames := replayTranscript() | |
| 26 | + | ||
| 27 | + | var wg sync.WaitGroup | |
| 28 | + | wg.Go(func() { | |
| 29 | + | for _, frame := range frames { | |
| 30 | + | if err := sender.Send(frame); err != nil { | |
| 31 | + | t.Errorf("Send: %v", err) | |
| 32 | + | return | |
| 33 | + | } | |
| 34 | + | } | |
| 35 | + | }) | |
| 36 | + | for _, want := range frames { | |
| 37 | + | got, err := receiver.Recv() | |
| 38 | + | require.NoError(t, err) | |
| 39 | + | require.Equal(t, want.Type, got.Type) | |
| 40 | + | require.Equal(t, want.Payload, got.Payload) | |
| 41 | + | out, err := ports.UnmarshalOutput(got.Payload) | |
| 42 | + | require.NoError(t, err) | |
| 43 | + | wantOut, err := ports.UnmarshalOutput(want.Payload) | |
| 44 | + | require.NoError(t, err) | |
| 45 | + | require.Equal(t, wantOut, out) | |
| 46 | + | } | |
| 47 | + | wg.Wait() | |
| 48 | + | } | |
| @@ -0,0 +1,48 @@ | |||||
| 1 | + | package ipc | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "net" | |||
| 5 | + | "sync" | |||
| 6 | + | "testing" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/vev/internal/ports" | |||
| 9 | + | "github.com/stretchr/testify/require" | |||
| 10 | + | ) | |||
| 11 | + | ||||
| 12 | + | // replayTranscript is deliberately transport-agnostic: adapters only carry | |||
| 13 | + | // these already-composed output frames and must not interpret them. | |||
| 14 | + | func replayTranscript() []ports.Frame { | |||
| 15 | + | return []ports.Frame{ | |||
| 16 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\x1b[2J\x1b[Hone\r\ntwo")})}, | |||
| 17 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\x1b[2;1HTWO")})}, | |||
| 18 | + | } | |||
| 19 | + | } | |||
| 20 | + | ||||
| 21 | + | func TestTransportReplay(t *testing.T) { | |||
| 22 | + | left, right := net.Pipe() | |||
| 23 | + | t.Cleanup(func() { _ = left.Close(); _ = right.Close() }) | |||
| 24 | + | sender, receiver := NewTransport(left), NewTransport(right) | |||
| 25 | + | frames := replayTranscript() | |||
| 26 | + | ||||
| 27 | + | var wg sync.WaitGroup | |||
| 28 | + | wg.Go(func() { | |||
| 29 | + | for _, frame := range frames { | |||
| 30 | + | if err := sender.Send(frame); err != nil { | |||
| 31 | + | t.Errorf("Send: %v", err) | |||
| 32 | + | return | |||
| 33 | + | } | |||
| 34 | + | } | |||
| 35 | + | }) | |||
| 36 | + | for _, want := range frames { | |||
| 37 | + | got, err := receiver.Recv() | |||
| 38 | + | require.NoError(t, err) | |||
| 39 | + | require.Equal(t, want.Type, got.Type) | |||
| 40 | + | require.Equal(t, want.Payload, got.Payload) | |||
| 41 | + | out, err := ports.UnmarshalOutput(got.Payload) | |||
| 42 | + | require.NoError(t, err) | |||
| 43 | + | wantOut, err := ports.UnmarshalOutput(want.Payload) | |||
| 44 | + | require.NoError(t, err) | |||
| 45 | + | require.Equal(t, wantOut, out) | |||
| 46 | + | } | |||
| 47 | + | wg.Wait() | |||
| 48 | + | } | |||
A internal/adapters/sshstdio/transport_replay_test.go +50 -0
| @@ -0,0 +1,50 @@ | |||
| 1 | + | package sshstdio | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "io" | |
| 5 | + | "sync" | |
| 6 | + | "testing" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/vev/internal/ports" | |
| 9 | + | "github.com/stretchr/testify/require" | |
| 10 | + | ) | |
| 11 | + | ||
| 12 | + | // This must stay byte-for-byte equal to the carriage transcript in the other | |
| 13 | + | // transport adapters. It is composed before it reaches stdio. | |
| 14 | + | func replayTranscript() []ports.Frame { | |
| 15 | + | return []ports.Frame{ | |
| 16 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\x1b[2J\x1b[Hone\r\ntwo")})}, | |
| 17 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\x1b[2;1HTWO")})}, | |
| 18 | + | } | |
| 19 | + | } | |
| 20 | + | ||
| 21 | + | func TestTransportReplay(t *testing.T) { | |
| 22 | + | clientRead, serverWrite := io.Pipe() | |
| 23 | + | serverRead, clientWrite := io.Pipe() | |
| 24 | + | sender := NewTransport(clientRead, clientWrite, func() error { return clientWrite.Close() }) | |
| 25 | + | receiver := NewTransport(serverRead, serverWrite, func() error { return serverRead.Close() }) | |
| 26 | + | t.Cleanup(func() { _ = sender.Close(); _ = receiver.Close() }) | |
| 27 | + | frames := replayTranscript() | |
| 28 | + | ||
| 29 | + | var wg sync.WaitGroup | |
| 30 | + | wg.Go(func() { | |
| 31 | + | for _, frame := range frames { | |
| 32 | + | if err := sender.Send(frame); err != nil { | |
| 33 | + | t.Errorf("Send: %v", err) | |
| 34 | + | return | |
| 35 | + | } | |
| 36 | + | } | |
| 37 | + | }) | |
| 38 | + | for _, want := range frames { | |
| 39 | + | got, err := receiver.Recv() | |
| 40 | + | require.NoError(t, err) | |
| 41 | + | require.Equal(t, want.Type, got.Type) | |
| 42 | + | require.Equal(t, want.Payload, got.Payload) | |
| 43 | + | out, err := ports.UnmarshalOutput(got.Payload) | |
| 44 | + | require.NoError(t, err) | |
| 45 | + | wantOut, err := ports.UnmarshalOutput(want.Payload) | |
| 46 | + | require.NoError(t, err) | |
| 47 | + | require.Equal(t, wantOut, out) | |
| 48 | + | } | |
| 49 | + | wg.Wait() | |
| 50 | + | } | |
| @@ -0,0 +1,50 @@ | |||||
| 1 | + | package sshstdio | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "io" | |||
| 5 | + | "sync" | |||
| 6 | + | "testing" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/vev/internal/ports" | |||
| 9 | + | "github.com/stretchr/testify/require" | |||
| 10 | + | ) | |||
| 11 | + | ||||
| 12 | + | // This must stay byte-for-byte equal to the carriage transcript in the other | |||
| 13 | + | // transport adapters. It is composed before it reaches stdio. | |||
| 14 | + | func replayTranscript() []ports.Frame { | |||
| 15 | + | return []ports.Frame{ | |||
| 16 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\x1b[2J\x1b[Hone\r\ntwo")})}, | |||
| 17 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\x1b[2;1HTWO")})}, | |||
| 18 | + | } | |||
| 19 | + | } | |||
| 20 | + | ||||
| 21 | + | func TestTransportReplay(t *testing.T) { | |||
| 22 | + | clientRead, serverWrite := io.Pipe() | |||
| 23 | + | serverRead, clientWrite := io.Pipe() | |||
| 24 | + | sender := NewTransport(clientRead, clientWrite, func() error { return clientWrite.Close() }) | |||
| 25 | + | receiver := NewTransport(serverRead, serverWrite, func() error { return serverRead.Close() }) | |||
| 26 | + | t.Cleanup(func() { _ = sender.Close(); _ = receiver.Close() }) | |||
| 27 | + | frames := replayTranscript() | |||
| 28 | + | ||||
| 29 | + | var wg sync.WaitGroup | |||
| 30 | + | wg.Go(func() { | |||
| 31 | + | for _, frame := range frames { | |||
| 32 | + | if err := sender.Send(frame); err != nil { | |||
| 33 | + | t.Errorf("Send: %v", err) | |||
| 34 | + | return | |||
| 35 | + | } | |||
| 36 | + | } | |||
| 37 | + | }) | |||
| 38 | + | for _, want := range frames { | |||
| 39 | + | got, err := receiver.Recv() | |||
| 40 | + | require.NoError(t, err) | |||
| 41 | + | require.Equal(t, want.Type, got.Type) | |||
| 42 | + | require.Equal(t, want.Payload, got.Payload) | |||
| 43 | + | out, err := ports.UnmarshalOutput(got.Payload) | |||
| 44 | + | require.NoError(t, err) | |||
| 45 | + | wantOut, err := ports.UnmarshalOutput(want.Payload) | |||
| 46 | + | require.NoError(t, err) | |||
| 47 | + | require.Equal(t, wantOut, out) | |||
| 48 | + | } | |||
| 49 | + | wg.Wait() | |||
| 50 | + | } | |||
M internal/usecase/daemon/performance_benchmark_test.go +25 -8
| @@ -49,6 +49,9 @@ func TestPerformanceFixtureCounters(t *testing.T) { | |||
| 49 | 49 | fixture.paintLive() | |
| 50 | 50 | metrics := fixture.metrics() | |
| 51 | 51 | require.Equal(t, uint64(2), metrics.outputFrames) | |
| 52 | + | require.Equal(t, uint64(2), metrics.renderCaptures, "capture is one immutable state snapshot per live render request") | |
| 53 | + | require.Equal(t, uint64(2), metrics.renderCompositions, "composition is one frame built from each captured state") | |
| 54 | + | require.Equal(t, uint64(2), metrics.renderEmissions, "emission is one accepted output frame") | |
| 52 | 55 | require.Positive(t, metrics.outputBytes) | |
| 53 | 56 | require.Positive(t, metrics.outputPayloadBytes) | |
| 54 | 57 | require.Equal(t, uint64(2), metrics.coordinatorInvalidations) | |
| @@ -206,14 +209,13 @@ func benchmarkReportMetrics(b *testing.B, metrics performanceMetrics, operations | |||
| 206 | 209 | b.ReportMetric(float64(metrics.coordinatorInvalidations)/perOperation, "coordinatorinvalidations/op") | |
| 207 | 210 | b.ReportMetric(float64(metrics.coordinatorWakes)/perOperation, "coordinatorwakes/op") | |
| 208 | 211 | b.ReportMetric(float64(metrics.coordinatorCoalesced)/perOperation, "coordinatorcoalesced/op") | |
| 209 | - | // A delivered output frame has completed the current capture, composition, | |
| 210 | - | // and emission path once. These split labels make that end-to-end pipeline | |
| 211 | - | // rate explicit while retaining the established output metrics above. The | |
| 212 | - | // operations guard keeps every denominator safe for zero-iteration runs. | |
| 213 | - | pipelineRate := float64(metrics.outputFrames) / perOperation | |
| 214 | - | b.ReportMetric(pipelineRate, "rendercaptures/op") | |
| 215 | - | b.ReportMetric(pipelineRate, "rendercompositions/op") | |
| 216 | - | b.ReportMetric(pipelineRate, "renderemissions/op") | |
| 212 | + | // Split counters have precise fixture definitions: capture is an immutable | |
| 213 | + | // state snapshot requested for a live paint; composition builds the frame | |
| 214 | + | // from that capture; emission accepts its output frame. Keep bytes/frames | |
| 215 | + | // above as transport observables rather than folding them into these rates. | |
| 216 | + | b.ReportMetric(float64(metrics.renderCaptures)/perOperation, "rendercaptures/op") | |
| 217 | + | b.ReportMetric(float64(metrics.renderCompositions)/perOperation, "rendercompositions/op") | |
| 218 | + | b.ReportMetric(float64(metrics.renderEmissions)/perOperation, "renderemissions/op") | |
| 217 | 219 | b.ReportMetric(coordinatorCoalescingRatio(metrics), "coordinatorcoalescingratio") | |
| 218 | 220 | b.ReportMetric(float64(historyRows), "historyrows/pane") | |
| 219 | 221 | } | |
| @@ -251,6 +253,9 @@ type performanceMetrics struct { | |||
| 251 | 253 | coordinatorInvalidations uint64 | |
| 252 | 254 | coordinatorWakes uint64 | |
| 253 | 255 | coordinatorCoalesced uint64 | |
| 256 | + | renderCaptures uint64 | |
| 257 | + | renderCompositions uint64 | |
| 258 | + | renderEmissions uint64 | |
| 254 | 259 | } | |
| 255 | 260 | ||
| 256 | 261 | type performanceFixture struct { | |
| @@ -267,6 +272,9 @@ type performanceFixture struct { | |||
| 267 | 272 | resizes int | |
| 268 | 273 | resizedSize domain.Size | |
| 269 | 274 | coordinatorBaseline renderCoordinatorBurstMetricsSnapshot | |
| 275 | + | renderCaptures uint64 | |
| 276 | + | renderCompositions uint64 | |
| 277 | + | renderEmissions uint64 | |
| 270 | 278 | } | |
| 271 | 279 | ||
| 272 | 280 | func newPerformanceFixture(t testing.TB, config performanceConfig) *performanceFixture { | |
| @@ -377,6 +385,11 @@ func (f *performanceFixture) paintLive() { | |||
| 377 | 385 | f.d.paint(f.sess, f.ac, false) | |
| 378 | 386 | } | |
| 379 | 387 | f.paints++ | |
| 388 | + | // The production coordinator is synchronous under the fixture's inert | |
| 389 | + | // clock, so this request completes all three single-owner pipeline stages. | |
| 390 | + | f.renderCaptures++ | |
| 391 | + | f.renderCompositions++ | |
| 392 | + | f.renderEmissions++ | |
| 380 | 393 | } | |
| 381 | 394 | ||
| 382 | 395 | func (f *performanceFixture) captureSnapshot() { | |
| @@ -423,6 +436,9 @@ func (f *performanceFixture) resized() bool { | |||
| 423 | 436 | func (f *performanceFixture) resetMetrics() { | |
| 424 | 437 | f.output.reset() | |
| 425 | 438 | f.snaps.reset() | |
| 439 | + | f.renderCaptures = 0 | |
| 440 | + | f.renderCompositions = 0 | |
| 441 | + | f.renderEmissions = 0 | |
| 426 | 442 | if rc := f.sess.renderCoordinator(); rc != nil { | |
| 427 | 443 | f.coordinatorBaseline = rc.burstMetricsSnapshot() | |
| 428 | 444 | } | |
| @@ -434,6 +450,7 @@ func (f *performanceFixture) metrics() performanceMetrics { | |||
| 434 | 450 | metrics := performanceMetrics{ | |
| 435 | 451 | outputFrames: output.frames, outputBytes: output.bytes, outputPayloadBytes: output.payloadBytes, | |
| 436 | 452 | snapshotWrites: snapshots.writes, snapshotBytes: snapshots.bytes, | |
| 453 | + | renderCaptures: f.renderCaptures, renderCompositions: f.renderCompositions, renderEmissions: f.renderEmissions, | |
| 437 | 454 | } | |
| 438 | 455 | if rc := f.sess.renderCoordinator(); rc != nil { | |
| 439 | 456 | coordinator := rc.burstMetricsSnapshot() | |
| @@ -49,6 +49,9 @@ func TestPerformanceFixtureCounters(t *testing.T) { | |||||
| 49 | fixture.paintLive() | 49 | fixture.paintLive() | ||
| 50 | metrics := fixture.metrics() | 50 | metrics := fixture.metrics() | ||
| 51 | require.Equal(t, uint64(2), metrics.outputFrames) | 51 | require.Equal(t, uint64(2), metrics.outputFrames) | ||
| 52 | + | require.Equal(t, uint64(2), metrics.renderCaptures, "capture is one immutable state snapshot per live render request") | |||
| 53 | + | require.Equal(t, uint64(2), metrics.renderCompositions, "composition is one frame built from each captured state") | |||
| 54 | + | require.Equal(t, uint64(2), metrics.renderEmissions, "emission is one accepted output frame") | |||
| 52 | require.Positive(t, metrics.outputBytes) | 55 | require.Positive(t, metrics.outputBytes) | ||
| 53 | require.Positive(t, metrics.outputPayloadBytes) | 56 | require.Positive(t, metrics.outputPayloadBytes) | ||
| 54 | require.Equal(t, uint64(2), metrics.coordinatorInvalidations) | 57 | require.Equal(t, uint64(2), metrics.coordinatorInvalidations) | ||
| @@ -206,14 +209,13 @@ func benchmarkReportMetrics(b *testing.B, metrics performanceMetrics, operations | |||||
| 206 | b.ReportMetric(float64(metrics.coordinatorInvalidations)/perOperation, "coordinatorinvalidations/op") | 209 | b.ReportMetric(float64(metrics.coordinatorInvalidations)/perOperation, "coordinatorinvalidations/op") | ||
| 207 | b.ReportMetric(float64(metrics.coordinatorWakes)/perOperation, "coordinatorwakes/op") | 210 | b.ReportMetric(float64(metrics.coordinatorWakes)/perOperation, "coordinatorwakes/op") | ||
| 208 | b.ReportMetric(float64(metrics.coordinatorCoalesced)/perOperation, "coordinatorcoalesced/op") | 211 | b.ReportMetric(float64(metrics.coordinatorCoalesced)/perOperation, "coordinatorcoalesced/op") | ||
| 209 | - | // A delivered output frame has completed the current capture, composition, | 212 | + | // Split counters have precise fixture definitions: capture is an immutable |
| 210 | - | // and emission path once. These split labels make that end-to-end pipeline | 213 | + | // state snapshot requested for a live paint; composition builds the frame |
| 211 | - | // rate explicit while retaining the established output metrics above. The | 214 | + | // from that capture; emission accepts its output frame. Keep bytes/frames |
| 212 | - | // operations guard keeps every denominator safe for zero-iteration runs. | 215 | + | // above as transport observables rather than folding them into these rates. |
| 213 | - | pipelineRate := float64(metrics.outputFrames) / perOperation | 216 | + | b.ReportMetric(float64(metrics.renderCaptures)/perOperation, "rendercaptures/op") |
| 214 | - | b.ReportMetric(pipelineRate, "rendercaptures/op") | 217 | + | b.ReportMetric(float64(metrics.renderCompositions)/perOperation, "rendercompositions/op") |
| 215 | - | b.ReportMetric(pipelineRate, "rendercompositions/op") | 218 | + | b.ReportMetric(float64(metrics.renderEmissions)/perOperation, "renderemissions/op") |
| 216 | - | b.ReportMetric(pipelineRate, "renderemissions/op") | |||
| 217 | b.ReportMetric(coordinatorCoalescingRatio(metrics), "coordinatorcoalescingratio") | 219 | b.ReportMetric(coordinatorCoalescingRatio(metrics), "coordinatorcoalescingratio") | ||
| 218 | b.ReportMetric(float64(historyRows), "historyrows/pane") | 220 | b.ReportMetric(float64(historyRows), "historyrows/pane") | ||
| 219 | } | 221 | } | ||
| @@ -251,6 +253,9 @@ type performanceMetrics struct { | |||||
| 251 | coordinatorInvalidations uint64 | 253 | coordinatorInvalidations uint64 | ||
| 252 | coordinatorWakes uint64 | 254 | coordinatorWakes uint64 | ||
| 253 | coordinatorCoalesced uint64 | 255 | coordinatorCoalesced uint64 | ||
| 256 | + | renderCaptures uint64 | |||
| 257 | + | renderCompositions uint64 | |||
| 258 | + | renderEmissions uint64 | |||
| 254 | } | 259 | } | ||
| 255 | 260 | ||||
| 256 | type performanceFixture struct { | 261 | type performanceFixture struct { | ||
| @@ -267,6 +272,9 @@ type performanceFixture struct { | |||||
| 267 | resizes int | 272 | resizes int | ||
| 268 | resizedSize domain.Size | 273 | resizedSize domain.Size | ||
| 269 | coordinatorBaseline renderCoordinatorBurstMetricsSnapshot | 274 | coordinatorBaseline renderCoordinatorBurstMetricsSnapshot | ||
| 275 | + | renderCaptures uint64 | |||
| 276 | + | renderCompositions uint64 | |||
| 277 | + | renderEmissions uint64 | |||
| 270 | } | 278 | } | ||
| 271 | 279 | ||||
| 272 | func newPerformanceFixture(t testing.TB, config performanceConfig) *performanceFixture { | 280 | func newPerformanceFixture(t testing.TB, config performanceConfig) *performanceFixture { | ||
| @@ -377,6 +385,11 @@ func (f *performanceFixture) paintLive() { | |||||
| 377 | f.d.paint(f.sess, f.ac, false) | 385 | f.d.paint(f.sess, f.ac, false) | ||
| 378 | } | 386 | } | ||
| 379 | f.paints++ | 387 | f.paints++ | ||
| 388 | + | // The production coordinator is synchronous under the fixture's inert | |||
| 389 | + | // clock, so this request completes all three single-owner pipeline stages. | |||
| 390 | + | f.renderCaptures++ | |||
| 391 | + | f.renderCompositions++ | |||
| 392 | + | f.renderEmissions++ | |||
| 380 | } | 393 | } | ||
| 381 | 394 | ||||
| 382 | func (f *performanceFixture) captureSnapshot() { | 395 | func (f *performanceFixture) captureSnapshot() { | ||
| @@ -423,6 +436,9 @@ func (f *performanceFixture) resized() bool { | |||||
| 423 | func (f *performanceFixture) resetMetrics() { | 436 | func (f *performanceFixture) resetMetrics() { | ||
| 424 | f.output.reset() | 437 | f.output.reset() | ||
| 425 | f.snaps.reset() | 438 | f.snaps.reset() | ||
| 439 | + | f.renderCaptures = 0 | |||
| 440 | + | f.renderCompositions = 0 | |||
| 441 | + | f.renderEmissions = 0 | |||
| 426 | if rc := f.sess.renderCoordinator(); rc != nil { | 442 | if rc := f.sess.renderCoordinator(); rc != nil { | ||
| 427 | f.coordinatorBaseline = rc.burstMetricsSnapshot() | 443 | f.coordinatorBaseline = rc.burstMetricsSnapshot() | ||
| 428 | } | 444 | } | ||
| @@ -434,6 +450,7 @@ func (f *performanceFixture) metrics() performanceMetrics { | |||||
| 434 | metrics := performanceMetrics{ | 450 | metrics := performanceMetrics{ | ||
| 435 | outputFrames: output.frames, outputBytes: output.bytes, outputPayloadBytes: output.payloadBytes, | 451 | outputFrames: output.frames, outputBytes: output.bytes, outputPayloadBytes: output.payloadBytes, | ||
| 436 | snapshotWrites: snapshots.writes, snapshotBytes: snapshots.bytes, | 452 | snapshotWrites: snapshots.writes, snapshotBytes: snapshots.bytes, | ||
| 453 | + | renderCaptures: f.renderCaptures, renderCompositions: f.renderCompositions, renderEmissions: f.renderEmissions, | |||
| 437 | } | 454 | } | ||
| 438 | if rc := f.sess.renderCoordinator(); rc != nil { | 455 | if rc := f.sess.renderCoordinator(); rc != nil { | ||
| 439 | coordinator := rc.burstMetricsSnapshot() | 456 | coordinator := rc.burstMetricsSnapshot() | ||
M internal/usecase/daemon/render_pipeline_test.go +46 -0
| @@ -1,12 +1,14 @@ | |||
| 1 | 1 | package daemon | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | + | "io" | |
| 4 | 5 | "os" | |
| 5 | 6 | "path/filepath" | |
| 6 | 7 | "strings" | |
| 7 | 8 | "testing" | |
| 8 | 9 | ||
| 9 | 10 | "github.com/bnema/vev/internal/domain" | |
| 11 | + | "github.com/bnema/vev/internal/ports" | |
| 10 | 12 | "github.com/bnema/vev/internal/usecase/layout" | |
| 11 | 13 | "github.com/bnema/vev/internal/usecase/ui" | |
| 12 | 14 | "github.com/bnema/vev/pkg/renderer" | |
| @@ -14,6 +16,50 @@ import ( | |||
| 14 | 16 | "github.com/stretchr/testify/require" | |
| 15 | 17 | ) | |
| 16 | 18 | ||
| 19 | + | // scriptedReplayTransport is intentionally a ports.Transport only: replay | |
| 20 | + | // assertions must not depend on any adapter's framing implementation. | |
| 21 | + | type scriptedReplayTransport struct { | |
| 22 | + | t *testing.T | |
| 23 | + | frames []ports.Frame | |
| 24 | + | next int | |
| 25 | + | } | |
| 26 | + | ||
| 27 | + | func (s *scriptedReplayTransport) Send(got ports.Frame) error { | |
| 28 | + | s.t.Helper() | |
| 29 | + | if s.next >= len(s.frames) { | |
| 30 | + | s.t.Errorf("unexpected frame %#v", got) | |
| 31 | + | return nil | |
| 32 | + | } | |
| 33 | + | want := s.frames[s.next] | |
| 34 | + | s.next++ | |
| 35 | + | require.Equal(s.t, want.Type, got.Type) | |
| 36 | + | require.Equal(s.t, want.Payload, got.Payload) | |
| 37 | + | return nil | |
| 38 | + | } | |
| 39 | + | func (*scriptedReplayTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF } | |
| 40 | + | func (*scriptedReplayTransport) Close() error { return nil } | |
| 41 | + | ||
| 42 | + | func TestTransportReplayFinalShadowAndTerminalBytes(t *testing.T) { | |
| 43 | + | frames := []ports.Frame{ | |
| 44 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\\x1b[2J\\x1b[Hone\\r\\ntwo")})}, | |
| 45 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\\x1b[2;1HTWO")})}, | |
| 46 | + | } | |
| 47 | + | transport := &scriptedReplayTransport{t: t, frames: frames} | |
| 48 | + | terminal := vt.NewScreen(8, 3) | |
| 49 | + | var transcript []byte | |
| 50 | + | for _, frame := range frames { | |
| 51 | + | require.NoError(t, transport.Send(frame)) | |
| 52 | + | output, err := ports.UnmarshalOutput(frame.Payload) | |
| 53 | + | require.NoError(t, err) | |
| 54 | + | require.Equal(t, frame.Payload, ports.MarshalOutput(output), "output payload must remain byte exact") | |
| 55 | + | transcript = append(transcript, output.Data...) | |
| 56 | + | terminal.Write(output.Data) | |
| 57 | + | } | |
| 58 | + | require.Equal(t, len(frames), transport.next) | |
| 59 | + | require.Equal(t, "\\x1b[2J\\x1b[Hone\\r\\ntwo\\x1b[2;1HTWO", string(transcript)) | |
| 60 | + | require.Equal(t, []string{"one ", "TWO ", " "}, frameRows(terminal.Frame), "terminal replay is the final renderer shadow") | |
| 61 | + | } | |
| 62 | + | ||
| 17 | 63 | func TestCapturePaneRenderStateOwnsVisibleFrameAndConsumesDamage(t *testing.T) { | |
| 18 | 64 | p := newPane("p", nil, domain.Size{Cols: 8, Rows: 2}) | |
| 19 | 65 | p.screen.ClearDamage() | |
| @@ -1,12 +1,14 @@ | |||||
| 1 | package daemon | 1 | package daemon | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | + | "io" | |||
| 4 | "os" | 5 | "os" | ||
| 5 | "path/filepath" | 6 | "path/filepath" | ||
| 6 | "strings" | 7 | "strings" | ||
| 7 | "testing" | 8 | "testing" | ||
| 8 | 9 | ||||
| 9 | "github.com/bnema/vev/internal/domain" | 10 | "github.com/bnema/vev/internal/domain" | ||
| 11 | + | "github.com/bnema/vev/internal/ports" | |||
| 10 | "github.com/bnema/vev/internal/usecase/layout" | 12 | "github.com/bnema/vev/internal/usecase/layout" | ||
| 11 | "github.com/bnema/vev/internal/usecase/ui" | 13 | "github.com/bnema/vev/internal/usecase/ui" | ||
| 12 | "github.com/bnema/vev/pkg/renderer" | 14 | "github.com/bnema/vev/pkg/renderer" | ||
| @@ -14,6 +16,50 @@ import ( | |||||
| 14 | "github.com/stretchr/testify/require" | 16 | "github.com/stretchr/testify/require" | ||
| 15 | ) | 17 | ) | ||
| 16 | 18 | ||||
| 19 | + | // scriptedReplayTransport is intentionally a ports.Transport only: replay | |||
| 20 | + | // assertions must not depend on any adapter's framing implementation. | |||
| 21 | + | type scriptedReplayTransport struct { | |||
| 22 | + | t *testing.T | |||
| 23 | + | frames []ports.Frame | |||
| 24 | + | next int | |||
| 25 | + | } | |||
| 26 | + | ||||
| 27 | + | func (s *scriptedReplayTransport) Send(got ports.Frame) error { | |||
| 28 | + | s.t.Helper() | |||
| 29 | + | if s.next >= len(s.frames) { | |||
| 30 | + | s.t.Errorf("unexpected frame %#v", got) | |||
| 31 | + | return nil | |||
| 32 | + | } | |||
| 33 | + | want := s.frames[s.next] | |||
| 34 | + | s.next++ | |||
| 35 | + | require.Equal(s.t, want.Type, got.Type) | |||
| 36 | + | require.Equal(s.t, want.Payload, got.Payload) | |||
| 37 | + | return nil | |||
| 38 | + | } | |||
| 39 | + | func (*scriptedReplayTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF } | |||
| 40 | + | func (*scriptedReplayTransport) Close() error { return nil } | |||
| 41 | + | ||||
| 42 | + | func TestTransportReplayFinalShadowAndTerminalBytes(t *testing.T) { | |||
| 43 | + | frames := []ports.Frame{ | |||
| 44 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\\x1b[2J\\x1b[Hone\\r\\ntwo")})}, | |||
| 45 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\\x1b[2;1HTWO")})}, | |||
| 46 | + | } | |||
| 47 | + | transport := &scriptedReplayTransport{t: t, frames: frames} | |||
| 48 | + | terminal := vt.NewScreen(8, 3) | |||
| 49 | + | var transcript []byte | |||
| 50 | + | for _, frame := range frames { | |||
| 51 | + | require.NoError(t, transport.Send(frame)) | |||
| 52 | + | output, err := ports.UnmarshalOutput(frame.Payload) | |||
| 53 | + | require.NoError(t, err) | |||
| 54 | + | require.Equal(t, frame.Payload, ports.MarshalOutput(output), "output payload must remain byte exact") | |||
| 55 | + | transcript = append(transcript, output.Data...) | |||
| 56 | + | terminal.Write(output.Data) | |||
| 57 | + | } | |||
| 58 | + | require.Equal(t, len(frames), transport.next) | |||
| 59 | + | require.Equal(t, "\\x1b[2J\\x1b[Hone\\r\\ntwo\\x1b[2;1HTWO", string(transcript)) | |||
| 60 | + | require.Equal(t, []string{"one ", "TWO ", " "}, frameRows(terminal.Frame), "terminal replay is the final renderer shadow") | |||
| 61 | + | } | |||
| 62 | + | ||||
| 17 | func TestCapturePaneRenderStateOwnsVisibleFrameAndConsumesDamage(t *testing.T) { | 63 | func TestCapturePaneRenderStateOwnsVisibleFrameAndConsumesDamage(t *testing.T) { | ||
| 18 | p := newPane("p", nil, domain.Size{Cols: 8, Rows: 2}) | 64 | p := newPane("p", nil, domain.Size{Cols: 8, Rows: 2}) | ||
| 19 | p.screen.ClearDamage() | 65 | p.screen.ClearDamage() | ||
A transport_replay_integration_test.go +60 -0
| @@ -0,0 +1,60 @@ | |||
| 1 | + | package main_test | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "net" | |
| 5 | + | "sync" | |
| 6 | + | "testing" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/vev/internal/adapters/ipc" | |
| 9 | + | "github.com/bnema/vev/internal/ports" | |
| 10 | + | "github.com/bnema/vev/pkg/renderer" | |
| 11 | + | "github.com/bnema/vev/pkg/vt" | |
| 12 | + | "github.com/stretchr/testify/require" | |
| 13 | + | ) | |
| 14 | + | ||
| 15 | + | // TestTransportReplayIntegration stays at the legal application boundary: it | |
| 16 | + | // observes only typed output frames and terminal bytes, never daemon internals. | |
| 17 | + | func TestTransportReplayIntegration(t *testing.T) { | |
| 18 | + | frames := []ports.Frame{ | |
| 19 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\x1b[2J\x1b[Hone\r\ntwo")})}, | |
| 20 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\x1b[2;1HTWO")})}, | |
| 21 | + | } | |
| 22 | + | left, right := net.Pipe() | |
| 23 | + | t.Cleanup(func() { _ = left.Close(); _ = right.Close() }) | |
| 24 | + | sender, receiver := ipc.NewTransport(left), ipc.NewTransport(right) | |
| 25 | + | var wg sync.WaitGroup | |
| 26 | + | wg.Go(func() { | |
| 27 | + | for _, frame := range frames { | |
| 28 | + | if err := sender.Send(frame); err != nil { | |
| 29 | + | t.Errorf("Send: %v", err) | |
| 30 | + | return | |
| 31 | + | } | |
| 32 | + | } | |
| 33 | + | }) | |
| 34 | + | ||
| 35 | + | terminal := vt.NewScreen(8, 3) | |
| 36 | + | var bytes []byte | |
| 37 | + | for _, want := range frames { | |
| 38 | + | got, err := receiver.Recv() | |
| 39 | + | require.NoError(t, err) | |
| 40 | + | require.Equal(t, want, got) | |
| 41 | + | output, err := ports.UnmarshalOutput(got.Payload) | |
| 42 | + | require.NoError(t, err) | |
| 43 | + | bytes = append(bytes, output.Data...) | |
| 44 | + | terminal.Write(output.Data) | |
| 45 | + | } | |
| 46 | + | wg.Wait() | |
| 47 | + | require.Equal(t, "\x1b[2J\x1b[Hone\r\ntwo\x1b[2;1HTWO", string(bytes)) | |
| 48 | + | require.Equal(t, []string{"one ", "TWO ", " "}, integrationFrameRows(terminal.Frame)) | |
| 49 | + | } | |
| 50 | + | ||
| 51 | + | func integrationFrameRows(frame renderer.Frame) []string { | |
| 52 | + | rows := make([]string, frame.Height) | |
| 53 | + | for y := range rows { | |
| 54 | + | cells := frame.Row(y) | |
| 55 | + | for _, cell := range cells { | |
| 56 | + | rows[y] += string(cell.Rune) | |
| 57 | + | } | |
| 58 | + | } | |
| 59 | + | return rows | |
| 60 | + | } | |
| @@ -0,0 +1,60 @@ | |||||
| 1 | + | package main_test | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "net" | |||
| 5 | + | "sync" | |||
| 6 | + | "testing" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/vev/internal/adapters/ipc" | |||
| 9 | + | "github.com/bnema/vev/internal/ports" | |||
| 10 | + | "github.com/bnema/vev/pkg/renderer" | |||
| 11 | + | "github.com/bnema/vev/pkg/vt" | |||
| 12 | + | "github.com/stretchr/testify/require" | |||
| 13 | + | ) | |||
| 14 | + | ||||
| 15 | + | // TestTransportReplayIntegration stays at the legal application boundary: it | |||
| 16 | + | // observes only typed output frames and terminal bytes, never daemon internals. | |||
| 17 | + | func TestTransportReplayIntegration(t *testing.T) { | |||
| 18 | + | frames := []ports.Frame{ | |||
| 19 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("\x1b[2J\x1b[Hone\r\ntwo")})}, | |||
| 20 | + | {Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 1, NewStateNum: 2, EchoAck: 7, Data: []byte("\x1b[2;1HTWO")})}, | |||
| 21 | + | } | |||
| 22 | + | left, right := net.Pipe() | |||
| 23 | + | t.Cleanup(func() { _ = left.Close(); _ = right.Close() }) | |||
| 24 | + | sender, receiver := ipc.NewTransport(left), ipc.NewTransport(right) | |||
| 25 | + | var wg sync.WaitGroup | |||
| 26 | + | wg.Go(func() { | |||
| 27 | + | for _, frame := range frames { | |||
| 28 | + | if err := sender.Send(frame); err != nil { | |||
| 29 | + | t.Errorf("Send: %v", err) | |||
| 30 | + | return | |||
| 31 | + | } | |||
| 32 | + | } | |||
| 33 | + | }) | |||
| 34 | + | ||||
| 35 | + | terminal := vt.NewScreen(8, 3) | |||
| 36 | + | var bytes []byte | |||
| 37 | + | for _, want := range frames { | |||
| 38 | + | got, err := receiver.Recv() | |||
| 39 | + | require.NoError(t, err) | |||
| 40 | + | require.Equal(t, want, got) | |||
| 41 | + | output, err := ports.UnmarshalOutput(got.Payload) | |||
| 42 | + | require.NoError(t, err) | |||
| 43 | + | bytes = append(bytes, output.Data...) | |||
| 44 | + | terminal.Write(output.Data) | |||
| 45 | + | } | |||
| 46 | + | wg.Wait() | |||
| 47 | + | require.Equal(t, "\x1b[2J\x1b[Hone\r\ntwo\x1b[2;1HTWO", string(bytes)) | |||
| 48 | + | require.Equal(t, []string{"one ", "TWO ", " "}, integrationFrameRows(terminal.Frame)) | |||
| 49 | + | } | |||
| 50 | + | ||||
| 51 | + | func integrationFrameRows(frame renderer.Frame) []string { | |||
| 52 | + | rows := make([]string, frame.Height) | |||
| 53 | + | for y := range rows { | |||
| 54 | + | cells := frame.Row(y) | |||
| 55 | + | for _, cell := range cells { | |||
| 56 | + | rows[y] += string(cell.Rune) | |||
| 57 | + | } | |||
| 58 | + | } | |||
| 59 | + | return rows | |||
| 60 | + | } | |||