From e988eab87e8f2e3384ef1f83459f1652883628e4 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Sun, 18 Oct 2026 18:04:21 +0800 Subject: [PATCH] B5.2: Add bounded split-output relay and source marks Relay separate stdout and stderr FIFOs through the existing ordered terminal writer. Public source marks identify exact logical byte ranges in the single raw output stream. Preserve interrupted output and resume each live reader after its complete admitted chunk is sent, including interruption at the final write boundary. Validation: all 15 repository and native checks pass on the current candidate, including Go race coverage and genuine Linux amd64 Bash 5.2.15, 5.2.37 and 5.3.9. A focused scheduling overlay verifies completed-write interruption for both streams and both delivery paths; the original implementation fails all four cases. Those deterministic cases skip without the overlay. ARM64 was compiled only. B5.3 owns production split start, resize and cleanup integration, including cleanup-before-keepalive closure, dual EOF, resource removal and final result barriers. This change supplies the isolated transport primitive. --- runtime/envoy/session/relay_linux.go | 38 +- runtime/envoy/session/split_relay_linux.go | 227 +++++ .../envoy/session/split_relay_linux_test.go | 884 ++++++++++++++++++ .../session/split_relay_native_linux_test.go | 219 +++++ runtime/envoy/session/startup_linux.go | 45 +- tests/envoy-split-streams/README.md | 41 + tests/envoy-split-streams/case-map.json | 60 ++ tests/envoy-split-streams/run.py | 49 + tests/test_native_proof_cleanup.py | 4 +- 9 files changed, 1560 insertions(+), 7 deletions(-) create mode 100644 runtime/envoy/session/split_relay_linux.go create mode 100644 runtime/envoy/session/split_relay_linux_test.go create mode 100644 runtime/envoy/session/split_relay_native_linux_test.go create mode 100644 tests/envoy-split-streams/README.md create mode 100644 tests/envoy-split-streams/case-map.json create mode 100644 tests/envoy-split-streams/run.py diff --git a/runtime/envoy/session/relay_linux.go b/runtime/envoy/session/relay_linux.go index a15d54b7..0fb4b97a 100644 --- a/runtime/envoy/session/relay_linux.go +++ b/runtime/envoy/session/relay_linux.go @@ -91,6 +91,7 @@ type Relay struct { phase context.Context writerCancel context.CancelCauseFunc reader *relayReader + split *splitRelay wake, done, readEnded, closing chan struct{} barrier, inputWriter chan struct{} err, readErr error @@ -130,6 +131,10 @@ func (r *Relay) Err() error { r.mu.Lock(); defer r.mu.Unlock(); func (r *Relay) ReadError() error { r.mu.Lock(); defer r.mu.Unlock(); return r.readErr } func (r *Relay) writeOutput(ctx context.Context, data []byte) (err error) { + return r.writeOutputStream(ctx, data, "pty") +} + +func (r *Relay) writeOutputStream(ctx context.Context, data []byte, stream string) (err error) { if len(data) == 0 { return nil } @@ -168,10 +173,16 @@ func (r *Relay) writeOutput(ctx context.Context, data []byte) (err error) { err = errors.Join(err, ctx.Err(), errWorkInterrupted) } if r.closeInterrupted(err) || (err != nil && errors.Is(context.Cause(ctx), errWorkInterrupted)) { - r.w.pendingPTY = append(r.w.pendingPTY, original[r.w.OutputThrough-before:]...) + pending := &r.w.pendingPTY + for i, name := range splitStreams { + if stream == name { + pending = &r.w.pendingSplit[i] + } + } + *pending = append(*pending, original[r.w.OutputThrough-before:]...) } }() - if err := r.w.mark(ctx, false); err != nil { + if err := r.w.markStream(ctx, stream, false); err != nil { return err } r.phaseMu.Lock() @@ -254,6 +265,16 @@ func (r *Relay) run() { for { r.mu.Lock() p := r.reader + s := r.split + var stdout, stderr <-chan struct{} + if s != nil { + if s.pumps[0] != nil { + stdout = s.pumps[0].wake + } + if s.pumps[1] != nil { + stderr = s.pumps[1].wake + } + } r.mu.Unlock() var output <-chan struct{} if p != nil { @@ -278,9 +299,22 @@ func (r *Relay) run() { r.readOutcome(errors.Join(r.err, r.w.channels.Err(), r.reader.err)) r.reader = nil } + if e := r.joinSplit(); e != nil { + r.err = errors.Join(r.err, e) + r.w.channels.fail(e) + } r.mu.Unlock() return case <-r.wake: + r.mu.Lock() + if r.ctx.Err() == nil { + r.flushSplitPending(r.w.channels.ctx) + } + r.mu.Unlock() + case <-stdout: + r.consumeSplit(s, 0) + case <-stderr: + r.consumeSplit(s, 1) case <-output: r.mu.Lock() if r.ctx.Err() != nil { diff --git a/runtime/envoy/session/split_relay_linux.go b/runtime/envoy/session/split_relay_linux.go new file mode 100644 index 00000000..9c1b0d26 --- /dev/null +++ b/runtime/envoy/session/split_relay_linux.go @@ -0,0 +1,227 @@ +//go:build linux + +package session + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "sync" + "syscall" + "time" +) + +var splitStreams = [2]string{"stdout", "stderr"} + +type splitRelay struct { + id string + files [2]*os.File + pumps [2]*relayReader + ended [2]bool + done chan struct{} // Independent read EOF facts, not output/result completion. + once sync.Once + stopIO func() +} + +// The setup owner retains its raw handles. Each poller owns one atomic +// close-on-exec duplicate of the validated reader, without reopening a path. +func (s *SplitResources) relayReaders(id string) (files [2]*os.File, err error) { + if s == nil || s.operation == nil || s.operation.id != id || len(s.operation.fifos) != 2 { + return files, fmt.Errorf("split reader operation mismatch") + } + if err = s.validate(); err != nil { + return files, err + } + defer func() { + if err != nil { + for _, file := range files { + if file != nil { + err = errors.Join(err, file.Close()) + } + } + } + }() + for i, fd := range s.operation.readers { + copyFD, _, errno := syscall.Syscall(syscall.SYS_FCNTL, uintptr(fd), syscall.F_DUPFD_CLOEXEC, 0) + if errno != 0 { + return files, errno + } + files[i] = os.NewFile(copyFD, splitStreams[i]) + var st syscall.Stat_t + if err = syscall.Fstat(int(copyFD), &st); err != nil { + return files, err + } + if !splitSame(st, s.operation.fifos[i].stat) { + return files, fmt.Errorf("split pump descriptor identity changed") + } + } + return files, s.validate() +} + +// attachSplit supplies the transport primitive. B5.3 owns its assembly into +// start, resize and cleanup; this method does not publish an operation result. +func (r *Relay) attachSplit(ctx context.Context, owner *SplitResources, id string) (err error) { + if err = ctx.Err(); err != nil { + return err + } + files, err := owner.relayReaders(id) + if err != nil { + return err + } + attached := false + defer func() { + if !attached { + for _, file := range files { + err = errors.Join(err, file.Close()) + } + } + }() + _, err = r.boundary(ctx, false, func(int64) error { + if r.split != nil { + return fmt.Errorf("split relay already attached") + } + s := &splitRelay{id: id, files: files, done: make(chan struct{})} + r.split, attached = s, true + s.stopIO = joinedAfterFunc(r.ctx, func() { + for _, file := range files { + _ = file.SetReadDeadline(time.Now()) + } + }) + for i, file := range files { + var e error + s.pumps[i], e = readRelay(file) + if e != nil { + return e + } + } + return nil + }) + return err +} + +func (r *Relay) splitOutcome(s *splitRelay, i int, err error) { + if errors.Is(err, io.EOF) { + s.ended[i] = true + if s.ended[0] && s.ended[1] { + s.once.Do(func() { close(s.done) }) + } + } else if err != nil && !r.closeInterrupted(err) { + r.err = errors.Join(r.err, err) + r.w.channels.fail(err) + r.cancel() + } +} + +func (r *Relay) splitWriteOutcome(err error) bool { + if err == nil { + return true + } + if !r.closeInterrupted(err) && !errors.Is(err, errWorkInterrupted) { + r.err = errors.Join(r.err, err) + r.w.channels.fail(err) + r.cancel() + } + return false +} + +// A split source may resume only after its complete admitted chunk reached the +// terminal writer. An interrupted write can still have an empty suffix when +// the boundary arrived after the final Write call; keep that source live in +// the same way as an uninterrupted write. A nonempty suffix remains parked. +func (r *Relay) resumeSplitIfReady(i int) { + s := r.split + if s == nil || s.pumps[i] == nil || len(r.w.pendingSplit[i]) != 0 || r.ctx.Err() != nil || r.w.channels.Err() != nil { + return + } + select { + case s.pumps[i].resume <- struct{}{}: + default: + } +} + +func (r *Relay) consumeSplit(s *splitRelay, i int) { + r.mu.Lock() + defer r.mu.Unlock() + if s != r.split || r.ctx.Err() != nil || s.pumps[i] == nil { + return + } + p := s.pumps[i] + data, err := p.take() + writeErr := r.writeOutputStream(r.w.channels.ctx, data, splitStreams[i]) + written := r.splitWriteOutcome(writeErr) + r.splitOutcome(s, i, err) + if err != nil { + <-p.done + s.pumps[i] = nil + } else if written || errors.Is(writeErr, errWorkInterrupted) { + r.resumeSplitIfReady(i) + } +} + +// A source with an undelivered suffix stays stopped. The existing boundary's +// wake resumes it only after that suffix reaches the sole terminal writer. +func (r *Relay) flushSplitPending(ctx context.Context) error { + for i, stream := range splitStreams { + data := r.w.pendingSplit[i] + if len(data) == 0 { + continue + } + r.w.pendingSplit[i] = nil + err := r.writeOutputStream(ctx, data, stream) + written := r.splitWriteOutcome(err) + if written || errors.Is(err, errWorkInterrupted) { + r.resumeSplitIfReady(i) + } + if !written { + return err + } + } + return nil +} + +// Called with mu held after relay cancellation or inside its phase boundary. +// It preserves admitted bytes; observing two EOFs is a separate caller proof. +func (r *Relay) joinSplit() (err error) { + s := r.split + if s == nil { + return nil + } + for i, p := range s.pumps { + if p != nil { + err = errors.Join(err, p.join(s.files[i])) + data, e := p.take() + r.w.pendingSplit[i] = append(r.w.pendingSplit[i], data...) + r.splitOutcome(s, i, e) + s.pumps[i] = nil + } + } + s.stopIO() + for i, file := range s.files { + if file != nil { + err = errors.Join(err, file.Close()) + s.files[i] = nil + } + } + return err +} + +// stopSplit joins only pump-owned I/O. The setup owner still owns its raw +// readers, writer keepalives and paths. This is not a cleanup/EOF barrier. +func (r *Relay) stopSplit(ctx context.Context) error { + _, err := r.boundary(ctx, false, func(int64) error { + if e := r.joinSplit(); e != nil { + return e + } + if e := r.flushSplitPending(ctx); e != nil { + return e + } + if r.err != nil { + return r.err + } + r.split = nil + return r.w.mark(ctx, true) + }) + return err +} diff --git a/runtime/envoy/session/split_relay_linux_test.go b/runtime/envoy/session/split_relay_linux_test.go new file mode 100644 index 00000000..564cad38 --- /dev/null +++ b/runtime/envoy/session/split_relay_linux_test.go @@ -0,0 +1,884 @@ +//go:build linux + +package session + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "runtime" + "sync" + "syscall" + "testing" + "time" + + "github.com/omry/omegaflow/runtime/envoy/protocol" +) + +func splitRelayFor(t *testing.T) (relayFixture, *splitTestOwner, [2]*os.File) { + t.Helper() + f := relayFixtureFor(t) + s := splitOwnerFor(t) + ctx := splitContext(t) + if _, _, err := s.Prepare(ctx, splitRequest()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = f.r.Close(splitContext(t)) + if err := s.Rollback(ctx, "operation-1"); err != nil { + t.Error(err) + } + }) + if err := f.r.attachSplit(ctx, s.SplitResources, "operation-1"); err != nil { + t.Fatal(err) + } + var writers [2]*os.File + for i, stream := range splitStreams { + file, err := os.OpenFile(filepath.Join(s.root, "split", "operation-1."+stream), os.O_WRONLY, 0) + if err != nil { + t.Fatal(err) + } + writers[i] = file + t.Cleanup(func() { _ = file.Close() }) + } + return f, s, writers +} + +func splitRelayManualFor(t *testing.T) (relayFixture, *splitTestOwner, [2]*os.File) { + t.Helper() + f := relayFixtureFor(t) + s := splitOwnerFor(t) + ctx := splitContext(t) + if _, _, err := s.Prepare(ctx, splitRequest()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = f.r.Close(splitContext(t)) + if err := s.Rollback(ctx, "operation-1"); err != nil { + t.Error(err) + } + }) + files, err := s.relayReaders("operation-1") + if err != nil { + t.Fatal(err) + } + attached := &splitRelay{id: "operation-1", files: files, done: make(chan struct{})} + attached.stopIO = joinedAfterFunc(f.r.ctx, func() { + for _, file := range files { + _ = file.SetReadDeadline(time.Now()) + } + }) + for i, file := range files { + attached.pumps[i], err = readRelay(file) + if err != nil { + t.Fatal(err) + } + } + f.r.mu.Lock() + f.r.split = attached + f.r.mu.Unlock() + select { + case f.r.wake <- struct{}{}: + default: + } + var writers [2]*os.File + for i, stream := range splitStreams { + file, err := os.OpenFile(filepath.Join(s.root, "split", "operation-1."+stream), os.O_WRONLY, 0) + if err != nil { + t.Fatal(err) + } + writers[i] = file + t.Cleanup(func() { _ = file.Close() }) + } + return f, s, writers +} + +func splitRead(t *testing.T, f relayFixture, n int) []byte { + t.Helper() + data := make([]byte, n) + _ = f.terminal.SetReadDeadline(time.Now().Add(3 * time.Second)) + if _, err := io.ReadFull(f.terminal, data); err != nil { + t.Fatal("raw output", n, err) + } + return data +} + +// The temporary focused overlay calls this seam immediately after a complete +// terminal write and before writeOutputStream samples its context. The normal +// source has no call site; the test skips there and the overlay run supplies +// the deterministic scheduling boundary needed to prove the interruption. +type splitWriteSeam struct { + stream string + entered chan struct{} + release chan struct{} + enteredOnce sync.Once + releaseOnce sync.Once +} + +var splitWriteSeamState struct { + sync.Mutex + current *splitWriteSeam +} + +func armSplitWriteSeam(stream string) *splitWriteSeam { + s := &splitWriteSeam{stream: stream, entered: make(chan struct{}), release: make(chan struct{})} + splitWriteSeamState.Lock() + splitWriteSeamState.current = s + splitWriteSeamState.Unlock() + return s +} + +func (s *splitWriteSeam) disarm() { + splitWriteSeamState.Lock() + if splitWriteSeamState.current == s { + splitWriteSeamState.current = nil + } + splitWriteSeamState.Unlock() + s.releaseOnce.Do(func() { close(s.release) }) +} + +func splitAfterCompletedStreamWrite(_ *Relay, stream string, ctx context.Context) { + splitWriteSeamState.Lock() + seam := splitWriteSeamState.current + if seam == nil || seam.stream != stream { + splitWriteSeamState.Unlock() + return + } + block := false + seam.enteredOnce.Do(func() { + close(seam.entered) + block = true + }) + splitWriteSeamState.Unlock() + if !block { + return + } + <-seam.release + <-ctx.Done() +} + +func waitSplitWriteSeam(t *testing.T, s *splitWriteSeam) bool { + t.Helper() + timer := time.NewTimer(500 * time.Millisecond) + defer timer.Stop() + select { + case <-s.entered: + return true + case <-timer.C: + return false + } +} + +func closeSplitSources(s *splitTestOwner, writers [2]*os.File) { + for i, writer := range writers { + _ = writer.Close() + if s.operation.keepalives[i] >= 0 { + _ = syscall.Close(s.operation.keepalives[i]) + s.operation.keepalives[i] = -1 + } + } +} + +// Interpret only the public mark stream over the raw log. There is no second +// production log or per-stream byte counter to serve as the reconstruction. +func splitReconstruct(t *testing.T, f relayFixture, raw []byte) map[string][]byte { + t.Helper() + result := map[string][]byte{} + stream, offset, elapsed := "pty", int64(0), int64(0) + for { + mark, ok := nextRelayFrame(t, f).(*protocol.OutputMark) + if !ok || mark.Offset < offset || mark.Offset > int64(len(raw)) || mark.ElapsedUS < elapsed { + t.Fatal("invalid stream frontier", mark, offset, len(raw)) + } + result[stream] = append(result[stream], raw[offset:mark.Offset]...) + stream, offset, elapsed = mark.Stream, mark.Offset, mark.ElapsedUS + if offset == int64(len(raw)) { + return result + } + } +} + +func TestSplitRelayExactLogicalBytesAndFirstSource(t *testing.T) { + for _, first := range []int{0, 1} { + t.Run(splitStreams[first], func(t *testing.T) { + f, _, writers := splitRelayFor(t) + var raw []byte + expected := map[string][]byte{} + for _, i := range []int{first, 1 - first, first, 2, 1 - first} { + data := []byte{'x', 0, 0xff, '\r', '\n', '\x1b', 3} + data = append(data, []byte("éλ終")...) + file, stream := f.slave, "pty" + if i < 2 { + file, stream = writers[i], splitStreams[i] + } + if _, err := file.Write(data); err != nil { + t.Fatal(err) + } + raw = append(raw, splitRead(t, f, len(data))...) + expected[stream] = append(expected[stream], data...) + } + if err := f.r.stopSplit(splitContext(t)); err != nil { + t.Fatal(err) + } + actual := splitReconstruct(t, f, raw) + for stream, want := range expected { + if !bytes.Equal(actual[stream], want) { + t.Fatalf("%s changed: %x != %x", stream, actual[stream], want) + } + } + // An empty boundary repeats the most recently written source. + if _, err := f.r.Drain(splitContext(t)); err != nil { + t.Fatal(err) + } + mark := nextRelayFrame(t, f).(*protocol.OutputMark) + if mark.Stream != splitStreams[1-first] || mark.Offset != int64(len(raw)) { + t.Fatal("empty boundary changed current source", mark) + } + }) + } +} + +func TestSplitRelayZeroBytesAndIndependentEOF(t *testing.T) { + f, s, writers := splitRelayFor(t) + f.r.mu.Lock() + attached := f.r.split + f.r.mu.Unlock() + for i := range writers { + _ = writers[i].Close() + if err := syscall.Close(s.operation.keepalives[i]); err != nil { + t.Fatal(err) + } + s.operation.keepalives[i] = -1 + if i == 0 { + wait := splitContext(t) + for { + f.r.mu.Lock() + ended := attached.ended[0] + f.r.mu.Unlock() + if ended { + break + } + if wait.Err() != nil { + t.Fatal("first source EOF was not observed") + } + runtime.Gosched() + } + // The other keepalive must prevent a false two-EOF assertion. + select { + case <-attached.done: + t.Fatal("one closed source completed both EOF facts") + default: + } + } + } + select { + case <-attached.done: + case <-splitContext(t).Done(): + t.Fatal("independent FIFO EOFs not observed") + } + if err := f.r.stopSplit(splitContext(t)); err != nil { + t.Fatal(err) + } + // Attachment and detachment both cover zero bytes, still attributed to PTY. + for range 3 { + m := nextRelayFrame(t, f).(*protocol.OutputMark) + if m.Offset != 0 || m.Stream != "pty" { + t.Fatal("zero-byte split invented a source", m) + } + } +} + +func TestSplitRelayConcurrentLargeStreams(t *testing.T) { + f, _, writers := splitRelayFor(t) + want := [2][]byte{bytes.Repeat([]byte("stdout\x00\xff\n"), 30000), bytes.Repeat([]byte("stderr\r\nλ"), 30000)} + done := make(chan error, 2) + for i, file := range writers { + go func() { _, err := file.Write(want[i]); done <- err }() + } + raw := splitRead(t, f, len(want[0])+len(want[1])) + for range writers { + if err := <-done; err != nil { + t.Fatal(err) + } + } + if err := f.r.stopSplit(splitContext(t)); err != nil { + t.Fatal(err) + } + got := splitReconstruct(t, f, raw) + for i, stream := range splitStreams { + if !bytes.Equal(got[stream], want[i]) { + t.Fatal("concurrent logical stream changed", stream, len(got[stream]), len(want[i])) + } + } +} + +func TestSplitRelayReaderIdentityAndOwnedDisposal(t *testing.T) { + f, s, _ := splitRelayFor(t) + f.r.mu.Lock() + var owned []int + var pumps []*relayReader + for i, file := range f.r.split.files { + raw, err := file.SyscallConn() + if err != nil { + f.r.mu.Unlock() + t.Fatal(err) + } + fd := -1 + if err := raw.Control(func(value uintptr) { fd = int(value) }); err != nil { + f.r.mu.Unlock() + t.Fatal(err) + } + owned = append(owned, fd) + pumps = append(pumps, f.r.split.pumps[i]) + flags, _, errno := syscall.Syscall(syscall.SYS_FCNTL, uintptr(fd), syscall.F_GETFD, 0) + if errno != 0 || flags&syscall.FD_CLOEXEC == 0 || fd == s.operation.readers[i] { + f.r.mu.Unlock() + t.Fatal("reader ownership/isolation", fd, flags, errno) + } + } + f.r.mu.Unlock() + if err := f.r.stopSplit(splitContext(t)); err != nil { + t.Fatal(err) + } + splitAssertClosed(t, owned) + for _, pump := range pumps { + select { + case <-pump.done: + default: + t.Fatal("retained pump") + } + } + if err := s.validate(); err != nil { + t.Fatal("detachment consumed setup-owned handles", err) + } + if err := f.r.attachSplit(splitContext(t), s.SplitResources, "other"); err == nil { + t.Fatal("wrong operation admitted") + } + ctx, cancel := context.WithCancel(splitContext(t)) + cancel() + if err := f.r.attachSplit(ctx, s.SplitResources, "operation-1"); !errors.Is(err, context.Canceled) { + t.Fatal("canceled attachment admitted", err) + } + if err := os.Chmod(filepath.Join(s.root, "split", "operation-1.stdout"), 0644); err != nil { + t.Fatal(err) + } + if err := f.r.attachSplit(splitContext(t), s.SplitResources, "operation-1"); err == nil { + t.Fatal("changed reader identity admitted") + } + if err := os.Chmod(filepath.Join(s.root, "split", "operation-1.stdout"), 0600); err != nil { + t.Fatal(err) + } +} + +func TestSplitRelayAdmittedChunksRemainBoundedAndSurviveClose(t *testing.T) { + f, _, writers := splitRelayFor(t) + f.r.mu.Lock() + s := f.r.split + for i, file := range writers { + if _, err := file.Write(bytes.Repeat([]byte{byte('a' + i)}, 4096)); err != nil { + f.r.mu.Unlock() + t.Fatal(err) + } + } + ctx := splitContext(t) + for _, p := range s.pumps { + for { + p.mu.Lock() + n := len(p.data) + p.mu.Unlock() + if n > 0 { + if n > 32768 { + t.Fatal("unbounded reader", n) + } + break + } + if ctx.Err() != nil { + f.r.mu.Unlock() + t.Fatal("no admitted chunk") + } + runtime.Gosched() + } + } + closed := make(chan error, 1) + go func() { closed <- f.r.Close(ctx) }() + <-f.r.ctx.Done() + f.r.mu.Unlock() + if err := <-closed; err != nil { + t.Fatal(err) + } + for i, pending := range f.r.w.pendingSplit { + if !bytes.Equal(pending, bytes.Repeat([]byte{byte('a' + i)}, 4096)) { + t.Fatal("lost admitted stream", i, len(pending)) + } + } + _ = f.slave.Close() + _ = f.r.w.PTY.SetReadDeadline(time.Now().Add(time.Second)) + if err := f.r.w.drainTerminal(ctx); err != nil { + t.Fatal(err) + } + got := splitReconstruct(t, f, splitRead(t, f, 8192)) + for i, stream := range splitStreams { + if !bytes.Equal(got[stream], bytes.Repeat([]byte{byte('a' + i)}, 4096)) { + t.Fatal("final drain misattributed retained output", stream) + } + } +} + +func TestSplitRelayReadFailureFailsSession(t *testing.T) { + f, _, _ := splitRelayFor(t) + f.r.mu.Lock() + file := f.r.split.files[0] + f.r.mu.Unlock() + if err := file.Close(); err != nil { + t.Fatal(err) + } + select { + case <-f.r.Done(): + case <-splitContext(t).Done(): + t.Fatal("read failure did not join the relay") + } + if f.r.Err() == nil || f.r.w.channels.Err() == nil { + t.Fatal("read failure produced success") + } +} + +func TestSplitRelayCadenceAndExistingSessionBudgets(t *testing.T) { + t.Run("cadence", func(t *testing.T) { + f, _, writers := splitRelayFor(t) + if _, err := writers[0].Write([]byte("first")); err != nil { + t.Fatal(err) + } + _ = splitRead(t, f, 5) + <-time.NewTimer(markCadence).C + if _, err := writers[0].Write([]byte("second")); err != nil { + t.Fatal(err) + } + _ = splitRead(t, f, 6) + if err := f.r.stopSplit(splitContext(t)); err != nil { + t.Fatal(err) + } + seen := false + for { + mark := nextRelayFrame(t, f).(*protocol.OutputMark) + if mark.Offset == 5 && mark.Stream == "stdout" { + seen = true + } + if mark.Offset == 11 { + break + } + } + if !seen { + t.Fatal("same-stream sender cadence mark omitted") + } + }) + for _, kind := range []string{"marks", "bytes"} { + t.Run(kind, func(t *testing.T) { + f, _, writers := splitRelayFor(t) + f.r.mu.Lock() + if kind == "marks" { + f.r.w.marks = markLimit + } else { + f.r.w.OutputThrough = outputLimit + } + before := f.r.w.OutputThrough + f.r.mu.Unlock() + if _, err := writers[1].Write([]byte("over budget")); err != nil { + t.Fatal(err) + } + select { + case <-f.r.Done(): + case <-splitContext(t).Done(): + t.Fatal("budget crossing did not fail relay") + } + if f.r.Err() == nil || f.r.w.OutputThrough != before { + t.Fatal("split source reset or crossed session budget", f.r.Err(), f.r.w.OutputThrough) + } + }) + } +} + +func TestSplitRelayRejectsDuplicateAttachment(t *testing.T) { + f, s, _ := splitRelayFor(t) + f.r.mu.Lock() + owned := f.r.split + f.r.mu.Unlock() + if err := f.r.attachSplit(splitContext(t), s.SplitResources, "operation-1"); err == nil { + t.Fatal("two split read owners admitted") + } + select { + case <-f.r.Done(): + case <-splitContext(t).Done(): + t.Fatal("failed attachment did not join partial ownership") + } + for _, file := range owned.files { + if file != nil { + t.Fatal("duplicate failure retained original pump file") + } + } + if err := s.validate(); err != nil { + t.Fatal("failed attachment consumed setup owner", err) + } +} + +func TestSplitRelayInterruptedWriterRetainsExactSourceSuffix(t *testing.T) { + for _, stream := range splitStreams { + t.Run(stream, func(t *testing.T) { + f := relayFixtureFor(t) + _ = f.r.w.channels.terminal.SetWriteBuffer(1024) + payload := bytes.Repeat([]byte("exact\x00\xff\r\n"), 300000) + ctx, cancel := context.WithCancelCause(splitContext(t)) + defer cancel(nil) + done := make(chan error, 1) + go func() { + f.r.mu.Lock() + err := f.r.writeOutputStream(ctx, payload, stream) + f.r.mu.Unlock() + done <- err + }() + // The source mark proves the writer entered its actual write phase. + m := nextRelayFrame(t, f).(*protocol.OutputMark) + if m.Stream != stream || m.Offset != 0 { + t.Fatal(m) + } + cancel(errWorkInterrupted) + if err := <-done; !errors.Is(err, errWorkInterrupted) { + t.Fatal("missing interrupted writer result", err) + } + i := 0 + if stream == "stderr" { + i = 1 + } + f.r.mu.Lock() + through := f.r.w.OutputThrough + pending := append([]byte(nil), f.r.w.pendingSplit[i]...) + f.r.mu.Unlock() + if len(pending) == 0 || !bytes.Equal(pending, payload[through:]) { + t.Fatal("interrupted suffix lost source/bytes", through, len(pending)) + } + read := make(chan error, 1) + go func() { + got := make([]byte, len(payload)) + _ = f.terminal.SetReadDeadline(time.Now().Add(3 * time.Second)) + _, err := io.ReadFull(f.terminal, got) + if err == nil && !bytes.Equal(got, payload) { + err = fmt.Errorf("interrupted bytes changed") + } + read <- err + }() + f.r.mu.Lock() + err := f.r.flushSplitPending(splitContext(t)) + f.r.mu.Unlock() + if e := <-read; err != nil || e != nil { + t.Fatal("retry failed", err, e) + } + }) + } +} + +func TestSplitRelayPhaseRetryResumesLiveSource(t *testing.T) { + f, _, writers := splitRelayFor(t) + if err := f.terminal.SetReadBuffer(4096); err != nil { + t.Fatal(err) + } + if err := f.r.w.channels.terminal.SetWriteBuffer(1024); err != nil { + t.Fatal(err) + } + f.r.mu.Lock() + _ = f.r.w.channels.terminal.SetWriteDeadline(time.Now().Add(100 * time.Millisecond)) + n, err := f.r.w.channels.terminal.Write(bytes.Repeat([]byte("p"), 1<<20)) + if err == nil || n == 1<<20 { + f.r.mu.Unlock() + t.Fatal("backpressure not established", n, err) + } + f.r.w.OutputThrough += int64(n) + f.r.mu.Unlock() + first, second := bytes.Repeat([]byte("a\x00\xff\n"), 1024), []byte("following FIFO chunk\r\n") + if _, err := writers[0].Write(first); err != nil { + t.Fatal(err) + } + for { + m := nextRelayFrame(t, f).(*protocol.OutputMark) + if m.Stream == "stdout" { + break + } + } + job, interrupt := context.WithCancelCause(splitContext(t)) + defer interrupt(nil) + boundary := make(chan error, 1) + go func() { _, err := f.r.boundary(job, false, nil); boundary <- err }() + wait := splitContext(t) + for { + f.r.phaseMu.Lock() + entered := f.r.phase == job + f.r.phaseMu.Unlock() + if entered { + break + } + if wait.Err() != nil { + t.Fatal("boundary did not acquire the blocked writer's epoch") + } + runtime.Gosched() + } + interrupt(errWorkInterrupted) + select { + case err := <-boundary: + if !errors.Is(err, context.Canceled) { + t.Fatal("wrong interruption", err) + } + case <-wait.Done(): + t.Fatal("boundary did not release") + } + if f.r.w.channels.Err() != nil { + t.Fatal("phase interruption failed session", f.r.w.channels.Err()) + } + want := append(bytes.Repeat([]byte("p"), n), first...) + want = append(want, second...) + read := make(chan error, 1) + go func() { + _ = f.terminal.SetReadDeadline(time.Now().Add(3 * time.Second)) + got := make([]byte, len(want)) + _, err := io.ReadFull(f.terminal, got) + if err == nil && !bytes.Equal(got, want) { + err = fmt.Errorf("live source changed after retry") + } + read <- err + }() + if _, err := writers[0].Write(second); err != nil { + t.Fatal(err) + } + if err := <-read; err != nil { + t.Fatal("live source did not resume after its retained suffix", err) + } + if err := f.r.stopSplit(splitContext(t)); err != nil { + t.Fatal(err) + } + if f.r.w.OutputThrough != int64(len(want)) || len(f.r.w.pendingSplit[0]) != 0 { + t.Fatal("retry lost or duplicated bytes") + } +} + +func TestSplitRelayCompletedInitialChunkInterruptionResumesLiveSource(t *testing.T) { + for source, stream := range splitStreams { + t.Run(stream, func(t *testing.T) { + seam := armSplitWriteSeam(stream) + t.Cleanup(seam.disarm) + f, s, writers := splitRelayFor(t) + first, second := []byte("first-complete-chunk"), []byte("next-fifo-chunk") + if _, err := writers[source].Write(first); err != nil { + t.Fatal(err) + } + if !waitSplitWriteSeam(t, seam) { + t.Skip("requires the temporary completed-write scheduling overlay") + } + if got := splitRead(t, f, len(first)); !bytes.Equal(got, first) { + t.Fatal("initial bytes changed", got) + } + + job, interrupt := context.WithCancelCause(splitContext(t)) + defer interrupt(nil) + boundary := make(chan error, 1) + go func() { _, err := f.r.boundary(job, false, nil); boundary <- err }() + wait := splitContext(t) + for { + f.r.phaseMu.Lock() + entered := f.r.phase == job + f.r.phaseMu.Unlock() + if entered { + break + } + if wait.Err() != nil { + t.Fatal("boundary did not acquire the completed writer epoch") + } + runtime.Gosched() + } + interrupt(errWorkInterrupted) + seam.releaseOnce.Do(func() { close(seam.release) }) + select { + case err := <-boundary: + if !errors.Is(err, context.Canceled) { + t.Fatal("wrong interruption", err) + } + case <-wait.Done(): + t.Fatal("boundary did not release") + } + + f.r.mu.Lock() + attached := f.r.split + pending, through := len(f.r.w.pendingSplit[source]), f.r.w.OutputThrough + f.r.mu.Unlock() + if pending != 0 || through != int64(len(first)) { + t.Fatalf("complete interrupted chunk was not acknowledged: pending=%d through=%d", pending, through) + } + if f.r.w.channels.Err() != nil { + t.Fatal("phase interruption failed session", f.r.w.channels.Err()) + } + if _, err := writers[source].Write(second); err != nil { + t.Fatal(err) + } + if got := splitRead(t, f, len(second)); !bytes.Equal(got, second) { + t.Fatal("resumed source bytes changed", got) + } + closeSplitSources(s, writers) + select { + case <-attached.done: + case <-splitContext(t).Done(): + f.r.mu.Lock() + ended := attached.ended + pumps := [2]bool{attached.pumps[0] != nil, attached.pumps[1] != nil} + f.r.mu.Unlock() + t.Fatalf("independent FIFO EOFs did not arrive: ended=%v pumps=%v channels=%v relay=%v", ended, pumps, f.r.w.channels.Err(), f.r.Err()) + } + if err := f.r.stopSplit(splitContext(t)); err != nil { + t.Fatal(err) + } + if f.r.w.OutputThrough != int64(len(first)+len(second)) || len(f.r.w.pendingSplit[source]) != 0 { + t.Fatal("resumed source duplicated or retained bytes") + } + }) + } +} + +func TestSplitRelayCompletedSuffixInterruptionResumesLiveSource(t *testing.T) { + for source, stream := range splitStreams { + t.Run(stream, func(t *testing.T) { + f, s, writers := splitRelayManualFor(t) + if err := f.r.w.channels.terminal.SetWriteBuffer(1024); err != nil { + t.Fatal(err) + } + prefix := bytes.Repeat([]byte("p"), 1<<20) + f.r.mu.Lock() + _ = f.r.w.channels.terminal.SetWriteDeadline(time.Now().Add(100 * time.Millisecond)) + n, err := f.r.w.channels.terminal.Write(prefix) + if err == nil || n == len(prefix) { + f.r.mu.Unlock() + t.Fatal("backpressure not established", n, err) + } + f.r.w.OutputThrough += int64(n) + f.r.mu.Unlock() + first := bytes.Repeat([]byte("source\x00\xff\r\n"), 3000) + job, interrupt := context.WithCancelCause(splitContext(t)) + defer interrupt(nil) + f.r.phaseMu.Lock() + f.r.phase = job + f.r.phaseMu.Unlock() + t.Cleanup(func() { + f.r.phaseMu.Lock() + f.r.phase = nil + f.r.phaseMu.Unlock() + }) + if _, err := writers[source].Write(first); err != nil { + t.Fatal(err) + } + for { + mark := nextRelayFrame(t, f).(*protocol.OutputMark) + if mark.Stream == stream { + break + } + } + interrupt(errWorkInterrupted) + wait := splitContext(t) + var through int64 + var pending []byte + for { + f.r.mu.Lock() + through = f.r.w.OutputThrough + pending = append([]byte(nil), f.r.w.pendingSplit[source]...) + f.r.mu.Unlock() + f.r.phaseMu.Lock() + writerDone := f.r.writerCancel == nil + f.r.phaseMu.Unlock() + if writerDone && len(pending) != 0 { + break + } + if wait.Err() != nil { + t.Fatal("interrupted source did not retain a suffix") + } + runtime.Gosched() + } + f.r.phaseMu.Lock() + f.r.phase = nil + f.r.phaseMu.Unlock() + before := int64(n) + if through < before || through-before > int64(len(first)) || !bytes.Equal(pending, first[through-before:]) { + t.Fatal("nonempty retained suffix was not preserved", through, len(pending)) + } + + seam := armSplitWriteSeam(stream) + t.Cleanup(seam.disarm) + retry, retryCancel := context.WithCancelCause(splitContext(t)) + defer retryCancel(nil) + flushed := make(chan error, 1) + go func() { + f.r.mu.Lock() + err := f.r.flushSplitPending(retry) + f.r.mu.Unlock() + flushed <- err + }() + read := make(chan struct { + data []byte + err error + }, 1) + go func() { + data := make([]byte, n+len(first)) + _ = f.terminal.SetReadDeadline(time.Now().Add(3 * time.Second)) + _, err := io.ReadFull(f.terminal, data) + read <- struct { + data []byte + err error + }{data, err} + }() + if !waitSplitWriteSeam(t, seam) { + t.Skip("requires the temporary completed-write scheduling overlay") + } + result := <-read + want := append(prefix[:n], first...) + if result.err != nil || !bytes.Equal(result.data, want) { + t.Fatal("retained suffix bytes changed", result.err) + } + retryCancel(errWorkInterrupted) + seam.releaseOnce.Do(func() { close(seam.release) }) + if err := <-flushed; !errors.Is(err, errWorkInterrupted) { + t.Fatal("missing interrupted flush result", err) + } + + f.r.mu.Lock() + attached := f.r.split + through = f.r.w.OutputThrough + pendingLen := len(f.r.w.pendingSplit[source]) + f.r.mu.Unlock() + if pendingLen != 0 || through != int64(n+len(first)) { + t.Fatalf("complete interrupted suffix was not acknowledged: pending=%d through=%d", pendingLen, through) + } + if f.r.w.channels.Err() != nil { + t.Fatal("suffix interruption failed session", f.r.w.channels.Err()) + } + following := []byte("following-fifo-chunk") + if _, err := writers[source].Write(following); err != nil { + t.Fatal(err) + } + if got := splitRead(t, f, len(following)); !bytes.Equal(got, following) { + t.Fatal("resumed suffix source bytes changed", got) + } + closeSplitSources(s, writers) + select { + case <-attached.done: + case <-splitContext(t).Done(): + f.r.mu.Lock() + ended := attached.ended + pumps := [2]bool{attached.pumps[0] != nil, attached.pumps[1] != nil} + f.r.mu.Unlock() + t.Fatalf("independent FIFO EOFs did not arrive after suffix retry: ended=%v pumps=%v channels=%v relay=%v", ended, pumps, f.r.w.channels.Err(), f.r.Err()) + } + if err := f.r.stopSplit(splitContext(t)); err != nil { + t.Fatal(err) + } + if f.r.w.OutputThrough != int64(n+len(first)+len(following)) || len(f.r.w.pendingSplit[source]) != 0 { + t.Fatal("suffix retry duplicated or retained bytes") + } + }) + } +} diff --git a/runtime/envoy/session/split_relay_native_linux_test.go b/runtime/envoy/session/split_relay_native_linux_test.go new file mode 100644 index 00000000..5a79fc92 --- /dev/null +++ b/runtime/envoy/session/split_relay_native_linux_test.go @@ -0,0 +1,219 @@ +//go:build linux + +package session + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "os" + "sync" + "syscall" + "testing" + "time" + + "github.com/omry/omegaflow/runtime/envoy/protocol" +) + +// Attach the real B5.2 transport to the real B5.1 resources at the test-only +// assembly boundary. Production start/completion integration belongs to B5.3. +type nativeSplitTransport struct { + *SplitResources + r *Relay + deadline time.Time +} + +func (s *nativeSplitTransport) Prepare(ctx context.Context, req protocol.Execute) (string, string, error) { + s.deadline, _ = ctx.Deadline() + out, errout, err := s.SplitResources.Prepare(ctx, req) + if err == nil { + err = s.r.attachSplit(ctx, s.SplitResources, req.OperationID) + } + return out, errout, err +} + +func TestNativeSplitStreamsExactBytes(t *testing.T) { + if os.Getenv("OMEGAFLOW_NATIVE_SPLIT_STREAMS") != "1" { + t.Skip("genuinely qualified isolated target required") + } + for _, first := range []string{"stdout", "stderr", "empty"} { + t.Run(first, func(t *testing.T) { + n := nativeOrdinaryFor(t) + owner := &nativeSplitTransport{SplitResources: NewSplitSetup(), r: n.c.r} + n.c.setup = owner + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _ = n.c.r.Close(ctx) + if owner.operation != nil { + original, end := context.WithDeadline(context.Background(), owner.deadline) + defer end() + if err := owner.Rollback(original, "operation-1"); err != nil { + t.Error("test-owned transport disposal", err) + } + } + }) + req := splitRequest() + req.Header = protocol.Header{Schema: protocol.Schema, Type: "execute", Seq: n.seq} + req.Source = ":" + want := map[string][]byte{} + if first != "empty" { + redirect := "" + if first == "stderr" { + redirect = " >&2" + } + req.Source = "builtin printf 'first\\000\\377\\r\\n'" + redirect + "; for i in {1..2048}; do builtin printf 'out-é\\000\\377\\r\\n'; builtin printf 'err-λ\\000\\376\\n' >&2; done" + want["stdout"] = bytes.Repeat([]byte("out-é\x00\xff\r\n"), 2048) + want["stderr"] = bytes.Repeat([]byte("err-λ\x00\xfe\n"), 2048) + want[first] = append([]byte("first\x00\xff\r\n"), want[first]...) + } + n.request(t, &req) + e := receiveInput(t, n.c.w.channels) + actual, ok := e.Message.(*protocol.Execute) + if !ok { + t.Fatal(e) + } + // Drain the actual terminal concurrently, so a blocked test reader + // cannot substitute a small-output assumption for transport evidence. + type rawResult struct { + data []byte + err error + } + rawDone := make(chan rawResult, 1) + sink := &nativeSplitSink{wake: make(chan struct{}, 1)} + _ = n.terminal.SetReadDeadline(time.Now().Add(5 * time.Second)) + go func() { + _, err := io.Copy(sink, n.terminal) + rawDone <- rawResult{sink.bytes(), err} + }() + rawJoined := false + t.Cleanup(func() { + if !rawJoined { + _ = n.terminal.SetReadDeadline(time.Now()) + select { + case <-rawDone: + case <-time.After(time.Second): + t.Error("test terminal collector did not join") + } + } + }) + out, err := n.c.Start(n.c.w.channels.ctx, *actual) + if err != nil || !out.Started { + t.Fatal("qualified split source did not start", out, err) + } + // The helper's proposal proves source return only. The test does not + // release input_closed or claim an operation result/descendant proof. + select { + case result := <-n.c.private.events: + proposal, ok := result.message.(*protocol.PrivateInputClose) + if result.err != nil || !ok || proposal.OperationID != req.OperationID { + t.Fatal("actual source return", result) + } + case result := <-n.c.private.fatal: + t.Fatal(result) + case <-time.After(5 * time.Second): + t.Fatal("actual source did not return") + } + n.c.r.mu.Lock() + attached := n.c.r.split + n.c.r.mu.Unlock() + // Only this source and the setup keepalives can write these FIFOs. + // B5.3 must supply its real cleanup ordering before doing this. + for i, fd := range owner.operation.keepalives { + if err := syscall.Close(fd); err != nil { + t.Fatal(err) + } + owner.operation.keepalives[i] = -1 + } + select { + case <-attached.done: + case <-time.After(3 * time.Second): + t.Fatal("actual FIFO streams did not end") + } + ctx := splitContext(t) + if err := n.c.r.stopSplit(ctx); err != nil { + t.Fatal(err) + } + through, err := n.c.r.Drain(ctx) + if err != nil { + t.Fatal(err) + } + for int64(len(sink.bytes())) < through-n.through { + select { + case <-sink.wake: + case <-ctx.Done(): + t.Fatal("terminal did not receive covering range") + } + } + _ = n.terminal.SetReadDeadline(time.Now()) + raw := <-rawDone + rawJoined = true + if !os.IsTimeout(raw.err) || int64(len(raw.data)) != through-n.through { + t.Fatal("raw native range", len(raw.data), through, n.through, raw.err) + } + got := map[string][]byte{} + stream, offset := "pty", n.through + started := false + for { + m := nativeFrame(t, n.reader) + encoded, err := protocol.EncodeEnvoy(m) + var h protocol.Header + if err != nil || json.Unmarshal(encoded, &h) != nil || h.Seq != n.lastSeq+1 { + t.Fatal("public sequence", m, err) + } + n.lastSeq = h.Seq + if mark, ok := m.(*protocol.OutputMark); ok { + if mark.Offset < offset || mark.Offset > through || mark.ElapsedUS < n.markElapsed { + t.Fatal("native mark frontier", mark) + } + got[stream] = append(got[stream], raw.data[offset-n.through:mark.Offset-n.through]...) + stream, offset, n.markElapsed = mark.Stream, mark.Offset, mark.ElapsedUS + if started && offset == through { + break + } + } else if event, ok := m.(*protocol.OperationStarted); !ok || event.OperationID != req.OperationID || event.OutputStart != out.OutputStart { + t.Fatal("unexpected native event", m) + } else { + started = true + } + } + for _, name := range splitStreams { + if !bytes.Equal(got[name], want[name]) { + t.Fatal("native logical bytes changed", name, len(got[name]), len(want[name])) + } + } + t.Logf("raw PTY controls: %q; exact stdout=%d stderr=%d", got["pty"], len(got["stdout"]), len(got["stderr"])) + // Readline's submit line ending is PTY traffic, outside both exact + // logical byte comparisons above. Keep its raw bytes in the evidence. + if visible := ordinaryVisible(string(got["pty"])); visible != "" && visible != "\n" { + t.Fatal("split source leaked into PTY", fmt.Sprintf("%q", visible)) + } + }) + } +} + +// A test-only terminal collector reports actual read progress, so stopping it +// cannot discard bytes that the writer sent but the socket has not delivered. +type nativeSplitSink struct { + mu sync.Mutex + buffer bytes.Buffer + wake chan struct{} +} + +func (s *nativeSplitSink) Write(p []byte) (int, error) { + s.mu.Lock() + n, err := s.buffer.Write(p) + s.mu.Unlock() + select { + case s.wake <- struct{}{}: + default: + } + return n, err +} +func (s *nativeSplitSink) bytes() []byte { + s.mu.Lock() + defer s.mu.Unlock() + return append([]byte(nil), s.buffer.Bytes()...) +} diff --git a/runtime/envoy/session/startup_linux.go b/runtime/envoy/session/startup_linux.go index 478ea322..9a871076 100644 --- a/runtime/envoy/session/startup_linux.go +++ b/runtime/envoy/session/startup_linux.go @@ -49,7 +49,9 @@ type Workload struct { closeErr error relayOwned atomic.Bool privateOwned atomic.Bool - pendingPTY []byte // at most one relay chunk, transferred after join + pendingPTY []byte // at most one relay chunk, transferred after join + pendingSplit [2][]byte // one admitted suffix per split source + outputStream string marks int lastMark time.Duration } @@ -486,23 +488,39 @@ func (w *Workload) verifyIdentity() error { } func (w *Workload) mark(ctx context.Context, force bool) error { + stream := w.outputStream + if stream == "" { + stream = "pty" + } + return w.markStream(ctx, stream, force) +} + +func (w *Workload) markStream(ctx context.Context, stream string, force bool) error { + current := w.outputStream + if current == "" { + current = "pty" + } + if stream != "pty" && stream != "stdout" && stream != "stderr" { + return fmt.Errorf("invalid output stream") + } elapsed := time.Since(w.ReadyAt) if elapsed < w.lastMark { return fmt.Errorf("output mark time regressed") } - if !force && elapsed-w.lastMark < markCadence { + if !force && stream == current && elapsed-w.lastMark < markCadence { return nil } if w.marks >= markLimit { return fmt.Errorf("output mark budget exhausted") } if err := w.channels.sendEvent(ctx, "output_mark", func(h protocol.Header) protocol.Message { - return protocol.OutputMark{Header: h, Offset: w.OutputThrough, Stream: "pty", ElapsedUS: elapsed.Microseconds()} + return protocol.OutputMark{Header: h, Offset: w.OutputThrough, Stream: stream, ElapsedUS: elapsed.Microseconds()} }); err != nil { return err } w.marks++ w.lastMark = elapsed + w.outputStream = stream return nil } @@ -614,6 +632,24 @@ func (w *Workload) shutdownIdle(phase context.Context, reason string, next func( // Called only after the selected shell and direct Awsh child have been reaped. func (w *Workload) drainTerminal(ctx context.Context) error { + for i, stream := range splitStreams { + if len(w.pendingSplit[i]) > 0 { + if err := w.markStream(ctx, stream, false); err != nil { + return err + } + before := w.OutputThrough + err := w.writeTerminal(ctx, w.pendingSplit[i], true) + w.pendingSplit[i] = w.pendingSplit[i][w.OutputThrough-before:] + if err != nil { + return err + } + } + } + if len(w.pendingPTY) > 0 { + if err := w.markStream(ctx, "pty", false); err != nil { + return err + } + } before := w.OutputThrough err := w.writeTerminal(ctx, w.pendingPTY, true) w.pendingPTY = w.pendingPTY[w.OutputThrough-before:] @@ -624,6 +660,9 @@ func (w *Workload) drainTerminal(ctx context.Context) error { for { n, e := w.PTY.Read(buffer[:]) if n > 0 { + if err := w.markStream(ctx, "pty", false); err != nil { + return err + } if err := w.writeTerminal(ctx, buffer[:n], true); err != nil { return err } diff --git a/tests/envoy-split-streams/README.md b/tests/envoy-split-streams/README.md new file mode 100644 index 00000000..1e8d9a56 --- /dev/null +++ b/tests/envoy-split-streams/README.md @@ -0,0 +1,41 @@ +# Split stream transport proof (B5.2) + +Run the portable isolated transport tests: + +```sh +cd runtime/envoy +go test -buildvcs=false -race ./session -run '^TestSplitRelay' -count=1 +``` + +Run actual Bash stream proofs from the repository root: + +```sh +python3 tests/envoy-split-streams/run.py --qualification QUALIFICATION --output NEW_DIRECTORY +``` + +The qualifier must contain genuine complete receipts for the three pinned +Linux amd64 targets. The runner checks exact Awsh and startup asset hashes, +stages the same session test binary in each immutable target, and retains +image identities, binary hashes, commands and native logs. Rebuildable binaries +are removed on success and failure. ARM64 compilation is not qualification. + +Each stream uses its own real FIFO and one bounded reader. Both feed the +existing sole terminal writer. Public output marks over the single global raw +offset reconstruct exact logical stdout and stderr. Tests compare binary, +invalid UTF-8, NUL, CR/LF and multibyte UTF-8 without normalization. Only the +known Readline controls in the separate PTY stream receive a diagnostic check; +the raw PTY bytes remain in the native log. + +Tests cover stdout-first and stderr-first output, concurrent large streams, +PTY interleaving, empty boundaries, independent EOF facts, reader identity and +close-on-exec isolation, joined disposal, source failure, bounded admission, +interrupted writes and exact attributed suffix recovery. An EOF fact alone does +not constitute an operation result or a cleanup receipt. + +The native test uses an explicit test-only attachment of the real setup and +relay primitives. It observes actual source return, then closes the test-owned +writer keepalives and joins the transport. It does not send input_closed or +claim completed split operation semantics. B5.3 owns production start/resize/ +completion integration and the cleanup-before-keepalive, dual-EOF/removal/result +barriers. B6 owns real descendant cleanup. The case map preserves these +cumulative owners; no approved corpus file is rewritten here. diff --git a/tests/envoy-split-streams/case-map.json b/tests/envoy-split-streams/case-map.json new file mode 100644 index 00000000..76cbaeeb --- /dev/null +++ b/tests/envoy-split-streams/case-map.json @@ -0,0 +1,60 @@ +{ + "slice": "B5.2", + "cases": [ + { + "inventory_id": "B1-C005-pty-start", + "test": "TestSplitRelayExactLogicalBytesAndFirstSource", + "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source", + "cumulative_owner": "B5.3", + "cumulative_closeout": false + }, + { + "inventory_id": "B1-C005-pty-repeat-current-stream", + "test": "TestSplitRelayExactLogicalBytesAndFirstSource", + "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source", + "cumulative_owner": "B5.3", + "cumulative_closeout": false + }, + { + "inventory_id": "B1-C005-stdout-start", + "test": "TestSplitRelayExactLogicalBytesAndFirstSource", + "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source", + "cumulative_owner": "B5.3", + "cumulative_closeout": false + }, + { + "inventory_id": "B1-C005-stdout-repeat-current-stream", + "test": "TestSplitRelayExactLogicalBytesAndFirstSource", + "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source", + "cumulative_owner": "B5.3", + "cumulative_closeout": false + }, + { + "inventory_id": "B1-C005-stderr-start", + "test": "TestSplitRelayExactLogicalBytesAndFirstSource", + "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source", + "cumulative_owner": "B5.3", + "cumulative_closeout": false + }, + { + "inventory_id": "B1-C005-stderr-repeat-current-stream", + "test": "TestSplitRelayExactLogicalBytesAndFirstSource", + "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source", + "cumulative_owner": "B5.3", + "cumulative_closeout": false + }, + { + "inventory_id": "B1-C034-interleaved-logical-bytes", + "test": "TestSplitRelayConcurrentLargeStreams", + "proof": "Exact binary and UTF-8 stdout/stderr reconstructed solely from public marks and one raw terminal log", + "cumulative_owner": "B6.3", + "cumulative_closeout": false + }, + { + "inventory_id": "B1-PACKAGE-B5.2", + "test": "TestNativeSplitStreamsExactBytes", + "proof": "Actual qualified Bash writes exact independent FIFO streams; stdout-first, stderr-first and zero-byte cases", + "qualification": "genuine pinned Linux amd64 targets only" + } + ] +} diff --git a/tests/envoy-split-streams/run.py b/tests/envoy-split-streams/run.py new file mode 100644 index 00000000..a1016796 --- /dev/null +++ b/tests/envoy-split-streams/run.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +"""Prove exact split stream bytes and public marks on genuinely qualified Bash targets.""" +import argparse +from pathlib import Path +import subprocess + +import sys + +# Reuse the existing proof-only staging boundary; production paths stay fixed. +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'envoy-session-startup')) +from staging import build_peer, save, sha, stage, verified_rows + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--qualification', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--release', choices=('5.2.15', '5.2.37', '5.3.9')) + args = parser.parse_args() + output = args.output.resolve() + output.mkdir(parents=True, exist_ok=False) + with build_peer(output, './session') as (binary, current_awsh): + results = [] + for row, entry, actual in verified_rows(args.qualification, current_awsh): + if args.release and args.release != row['release']: + continue + iid, context = stage(output, row['release'], entry, actual, binary) + with (context / 'native.log').open('w') as log: + result = subprocess.run( + ['docker', 'run', '--rm', '--network=none', + '-e', 'OMEGAFLOW_NATIVE_SPLIT_STREAMS=1', + '--entrypoint=/omegaflow-runtime/bin/envoy', iid, + '-test.run=^TestNativeSplitStreams', '-test.v'], + stdout=log, stderr=subprocess.STDOUT, timeout=45) + results.append({'release': row['release'], 'qualified_awsh_sha256': current_awsh, + 'receipt_sha256': row['receipt_sha256'], 'native_image': iid, + 'test_binary_sha256': sha(binary), 'returncode': result.returncode, + 'log_path': str(context / 'native.log'), + 'log_sha256': sha(context / 'native.log')}) + save(output / 'results.json', results) + if result.returncode: + raise ValueError('actual split stream proof failed: ' + str(context / 'native.log')) + if not results: + raise ValueError('no selected qualified target was exercised') + print('Native split stream evidence: ' + str(output)) + + +if __name__ == '__main__': + main() diff --git a/tests/test_native_proof_cleanup.py b/tests/test_native_proof_cleanup.py index 88c68f0b..1cd8a319 100644 --- a/tests/test_native_proof_cleanup.py +++ b/tests/test_native_proof_cleanup.py @@ -10,11 +10,11 @@ import pytest PROOFS = Path(__file__).parent / 'envoy-session-startup' -RUNNERS = ('private', 'public', 'ordinary', 'resize', 'lifecycle', 'gate', 'controls', 'split-setup') +RUNNERS = ('private', 'public', 'ordinary', 'resize', 'lifecycle', 'gate', 'controls', 'split-setup', 'split-streams') def load(name): - path = PROOFS.parent / 'envoy-split-setup/run.py' if name == 'split-setup' else PROOFS / (name + '.py') + path = PROOFS.parent / ('envoy-' + name + '/run.py') if name in ('split-setup', 'split-streams') else PROOFS / (name + '.py') spec = importlib.util.spec_from_file_location('proof_' + name, path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module)