detached 4c975c7

test(resilience): add visual QUIC/SSH impairment harness

Userspace UDP/TCP relays, NeferWL headless + foot captures, pixel comparator and semantic checks for degraded links, blackouts, cancellation and held-input replay. Run with make remote-resilience-visual.
commit
4c975c70715b33fa42fd960b68b6705795d1a6f5
author
brice · · 2026-10-07 05:35 UTC
parent
1999182
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
11 files · +1038 -2
Unified
UnifiedSplit
M Makefile +4 -0
@@ -31,6 +31,10 @@ mocks:
3131 remote-acceptance:
3232 scripts/remote-picker-harness/run.sh
3333
34+.PHONY: remote-resilience-visual
35+remote-resilience-visual:
36+ python3 scripts/remote-resilience-harness/visual.py
37+
3438 # protocol regenerates the Protobuf wire contract from
3539 # internal/protocol/wire/schema using the pinned Go tools (buf + protoc-gen-go).
3640 protocol:
@@ -31,6 +31,10 @@ mocks:
31 remote-acceptance:31 remote-acceptance:
32 scripts/remote-picker-harness/run.sh32 scripts/remote-picker-harness/run.sh
33 33
34+.PHONY: remote-resilience-visual
35+remote-resilience-visual:
36+ python3 scripts/remote-resilience-harness/visual.py
37+
34 # protocol regenerates the Protobuf wire contract from38 # protocol regenerates the Protobuf wire contract from
35 # internal/protocol/wire/schema using the pinned Go tools (buf + protoc-gen-go).39 # internal/protocol/wire/schema using the pinned Go tools (buf + protoc-gen-go).
36 protocol:40 protocol:
M docs/remote-resilience.md +3 -1
@@ -30,6 +30,8 @@ The transport is fixed when the host is added (`quic` by default). To switch, ru
3030 - If the reconnect window has expired but the session still exists, the client opens a fresh attachment to it.
3131 - Several clients can attach to the same session. Each keeps its own window, tab, focus, and copy mode.
3232
33+The command palette is rendered by the remote daemon, so opening it needs a working link. A client-owned session picker that is already open remains locally interactive during an outage.
34+
3335 ## Connection states
3436
3537 | State | Meaning |
@@ -79,6 +81,6 @@ VEV_LOG=debug vev attach user@host
7981 scripts/debug-remote-attach.sh user@host # collects redacted connection health
8082 ```
8183
82-Useful manual checks: block UDP, switch networks or VPN, suspend the laptop, then reattach to the same session.
84+Useful manual checks: block UDP, switch networks or VPN, suspend the laptop, then reattach to the same session. For isolated QUIC/SSH impairment and real Wayland terminal captures, see the [visual resilience harness](../scripts/remote-resilience-harness/README.md).
8385
8486 Keep hostnames, usernames, and keys out of shared logs and screenshots.
@@ -30,6 +30,8 @@ The transport is fixed when the host is added (`quic` by default). To switch, ru
30 - If the reconnect window has expired but the session still exists, the client opens a fresh attachment to it.30 - If the reconnect window has expired but the session still exists, the client opens a fresh attachment to it.
31 - Several clients can attach to the same session. Each keeps its own window, tab, focus, and copy mode.31 - Several clients can attach to the same session. Each keeps its own window, tab, focus, and copy mode.
32 32
33+The command palette is rendered by the remote daemon, so opening it needs a working link. A client-owned session picker that is already open remains locally interactive during an outage.
34+
33 ## Connection states35 ## Connection states
34 36
35 | State | Meaning |37 | State | Meaning |
@@ -79,6 +81,6 @@ VEV_LOG=debug vev attach user@host
79 scripts/debug-remote-attach.sh user@host # collects redacted connection health81 scripts/debug-remote-attach.sh user@host # collects redacted connection health
80 ```82 ```
81 83
82-Useful manual checks: block UDP, switch networks or VPN, suspend the laptop, then reattach to the same session.84+Useful manual checks: block UDP, switch networks or VPN, suspend the laptop, then reattach to the same session. For isolated QUIC/SSH impairment and real Wayland terminal captures, see the [visual resilience harness](../scripts/remote-resilience-harness/README.md).
83 85
84 Keep hostnames, usernames, and keys out of shared logs and screenshots.86 Keep hostnames, usernames, and keys out of shared logs and screenshots.
M scripts/demo/Dockerfile +1 -1
@@ -7,7 +7,7 @@ RUN --mount=type=cache,target=/var/cache/pacman/pkg,sharing=locked \
77 vhs ttyd ffmpeg chromium \
88 htop fastfetch gifsicle imagemagick \
99 eza fzf neovim man-db man-pages coreutils \
10- ncurses openssh util-linux \
10+ ncurses openssh util-linux python \
1111 ttf-jetbrains-mono-nerd
1212 RUN useradd -m -u 1000 demo
1313 # P8.2 uses the exact worktree binary built by sandbox-run.sh, never a
@@ -7,7 +7,7 @@ RUN --mount=type=cache,target=/var/cache/pacman/pkg,sharing=locked \
7 vhs ttyd ffmpeg chromium \7 vhs ttyd ffmpeg chromium \
8 htop fastfetch gifsicle imagemagick \8 htop fastfetch gifsicle imagemagick \
9 eza fzf neovim man-db man-pages coreutils \9 eza fzf neovim man-db man-pages coreutils \
10- ncurses openssh util-linux \10+ ncurses openssh util-linux python \
11 ttf-jetbrains-mono-nerd11 ttf-jetbrains-mono-nerd
12 RUN useradd -m -u 1000 demo12 RUN useradd -m -u 1000 demo
13 # P8.2 uses the exact worktree binary built by sandbox-run.sh, never a13 # P8.2 uses the exact worktree binary built by sandbox-run.sh, never a
A scripts/remote-resilience-harness/README.md +25 -0
@@ -0,0 +1,25 @@
1+# Visual remote resilience acceptance
2+
3+`make remote-resilience-visual` runs a disposable remote session in a real foot terminal under NeferWL headless. Physical keys enter through the compositor. PNG differences and passive vev captures jointly verify local picker response within two seconds. A stopped client is the negative control.
4+
5+Prerequisites: Go 1.27, rootless Docker, Python 3, ssh-keygen, foot, and a GPU supported by NeferWL. Set `NEFERWL_SOURCE` to the compositor checkout or `NEFERWL_BIN` to its built binary. No desktop Wayland socket, user credentials or host network configuration is shared with containers.
6+
7+```sh
8+make remote-resilience-visual
9+VEV_RESILIENCE_TRANSPORT=ssh make remote-resilience-visual
10+VEV_RESILIENCE_DEGRADED=1 make remote-resilience-visual
11+VEV_RESILIENCE_TRANSPORT=ssh VEV_RESILIENCE_DEGRADED=1 make remote-resilience-visual
12+VEV_RESILIENCE_LINK=udp make remote-resilience-visual
13+VEV_RESILIENCE_LINK=tcp make remote-resilience-visual
14+VEV_RESILIENCE_LONG=1 make remote-resilience-visual
15+VEV_RESILIENCE_LONG=1 VEV_RESILIENCE_RESUME=1 make remote-resilience-visual
16+VEV_RESILIENCE_LONG=1 VEV_RESILIENCE_CANCEL=Escape make remote-resilience-visual
17+```
18+
19+QUIC is routed through the existing bounded seeded UDP simulator. A fixture-only SSH wrapper preserves authentication fields and rewrites the dynamic bootstrap port to its relay. SSH runs through a TCP relay. The degraded profile applies 150 ms delay, up to 50 ms jitter and 5% datagram loss to UDP; TCP waits 150 ms per forwarded read and is paced to 64 KiB/s, without dropping stream bytes; many small reads therefore add up to more than 150 ms. Both transports can be blacked out together without privileges.
20+
21+The short scenario opens the client picker while connected, edits it during blackout, restores the route and checks the named session identity and recovery output. Opening the daemon-rendered palette during blackout is not a local responsiveness guarantee. The optional long scenario asserts actual Connecting state and Ctrl-C or Esc cancellation. It cannot tell backoff from a blocked stream open; fake-clock client tests cover cancellation during stream opening. It fails if cancellation cannot reach the resume owner.
22+
23+Artifacts are printed at exit: captures, pixel counts, observation contexts, relay counters and fixture logs. Containers, image, network and temporary credentials are removed. Unsupported rendering and failed assertions exit nonzero.
24+
25+This is not a mobile-network emulator: TCP impairments model stream backpressure, not packet retransmissions; IP migration, CGNAT and exactly-once replay are not validated. `VEV_RESILIENCE_LINK=udp|tcp` isolates short asymmetric blackouts. `VEV_RESILIENCE_RESUME=1` restores a long outage and checks two held fixture commands in order. Detach timing is covered by fake-clock client tests, not this visual entrypoint.
@@ -0,0 +1,25 @@
1+# Visual remote resilience acceptance
2+
3+`make remote-resilience-visual` runs a disposable remote session in a real foot terminal under NeferWL headless. Physical keys enter through the compositor. PNG differences and passive vev captures jointly verify local picker response within two seconds. A stopped client is the negative control.
4+
5+Prerequisites: Go 1.27, rootless Docker, Python 3, ssh-keygen, foot, and a GPU supported by NeferWL. Set `NEFERWL_SOURCE` to the compositor checkout or `NEFERWL_BIN` to its built binary. No desktop Wayland socket, user credentials or host network configuration is shared with containers.
6+
7+```sh
8+make remote-resilience-visual
9+VEV_RESILIENCE_TRANSPORT=ssh make remote-resilience-visual
10+VEV_RESILIENCE_DEGRADED=1 make remote-resilience-visual
11+VEV_RESILIENCE_TRANSPORT=ssh VEV_RESILIENCE_DEGRADED=1 make remote-resilience-visual
12+VEV_RESILIENCE_LINK=udp make remote-resilience-visual
13+VEV_RESILIENCE_LINK=tcp make remote-resilience-visual
14+VEV_RESILIENCE_LONG=1 make remote-resilience-visual
15+VEV_RESILIENCE_LONG=1 VEV_RESILIENCE_RESUME=1 make remote-resilience-visual
16+VEV_RESILIENCE_LONG=1 VEV_RESILIENCE_CANCEL=Escape make remote-resilience-visual
17+```
18+
19+QUIC is routed through the existing bounded seeded UDP simulator. A fixture-only SSH wrapper preserves authentication fields and rewrites the dynamic bootstrap port to its relay. SSH runs through a TCP relay. The degraded profile applies 150 ms delay, up to 50 ms jitter and 5% datagram loss to UDP; TCP waits 150 ms per forwarded read and is paced to 64 KiB/s, without dropping stream bytes; many small reads therefore add up to more than 150 ms. Both transports can be blacked out together without privileges.
20+
21+The short scenario opens the client picker while connected, edits it during blackout, restores the route and checks the named session identity and recovery output. Opening the daemon-rendered palette during blackout is not a local responsiveness guarantee. The optional long scenario asserts actual Connecting state and Ctrl-C or Esc cancellation. It cannot tell backoff from a blocked stream open; fake-clock client tests cover cancellation during stream opening. It fails if cancellation cannot reach the resume owner.
22+
23+Artifacts are printed at exit: captures, pixel counts, observation contexts, relay counters and fixture logs. Containers, image, network and temporary credentials are removed. Unsupported rendering and failed assertions exit nonzero.
24+
25+This is not a mobile-network emulator: TCP impairments model stream backpressure, not packet retransmissions; IP migration, CGNAT and exactly-once replay are not validated. `VEV_RESILIENCE_LINK=udp|tcp` isolates short asymmetric blackouts. `VEV_RESILIENCE_RESUME=1` restores a long outage and checks two held fixture commands in order. Detach timing is covered by fake-clock client tests, not this visual entrypoint.
A scripts/remote-resilience-harness/pixels/main.go +77 -0
@@ -0,0 +1,77 @@
1+// Command pixels compares the fixed 1280x720 fixture's terminal content.
2+package main
3+
4+import (
5+ "encoding/json"
6+ "errors"
7+ "fmt"
8+ "image"
9+ _ "image/png"
10+ "os"
11+)
12+
13+const fixtureMargin = 100
14+const minimumChangedPixels = 20
15+
16+func load(path string) (image.Image, error) {
17+ f, err := os.Open(path)
18+ if err != nil {
19+ return nil, err
20+ }
21+ defer f.Close()
22+ im, _, err := image.Decode(f)
23+ return im, err
24+}
25+
26+func compare(a, b image.Image) (int, error) {
27+ if a.Bounds() != b.Bounds() {
28+ return 0, errors.New("capture dimensions changed")
29+ }
30+ bounds := a.Bounds()
31+ if bounds.Dx() <= 2*fixtureMargin || bounds.Dy() <= 2*fixtureMargin {
32+ return 0, errors.New("capture has no comparison region")
33+ }
34+ changed := 0
35+ for y := bounds.Min.Y + fixtureMargin; y < bounds.Max.Y-fixtureMargin; y++ {
36+ for x := bounds.Min.X + fixtureMargin; x < bounds.Max.X-fixtureMargin; x++ {
37+ ar, ag, ab, _ := a.At(x, y).RGBA()
38+ br, bg, bb, _ := b.At(x, y).RGBA()
39+ if ar != br || ag != bg || ab != bb {
40+ changed++
41+ }
42+ }
43+ }
44+ return changed, nil
45+}
46+
47+func execute(args []string) (int, error) {
48+ if len(args) != 2 {
49+ return 2, errors.New("usage: pixels before.png after.png")
50+ }
51+ a, err := load(args[0])
52+ if err != nil {
53+ return 2, err
54+ }
55+ b, err := load(args[1])
56+ if err != nil {
57+ return 2, err
58+ }
59+ changed, err := compare(a, b)
60+ if err != nil {
61+ return 2, err
62+ }
63+ if err := json.NewEncoder(os.Stdout).Encode(map[string]int{"changed_pixels": changed}); err != nil {
64+ return 2, err
65+ }
66+ if changed < minimumChangedPixels {
67+ return 1, nil
68+ }
69+ return 0, nil
70+}
71+func main() {
72+ code, err := execute(os.Args[1:])
73+ if err != nil {
74+ fmt.Fprintln(os.Stderr, err)
75+ }
76+ os.Exit(code)
77+}
@@ -0,0 +1,77 @@
1+// Command pixels compares the fixed 1280x720 fixture's terminal content.
2+package main
3+
4+import (
5+ "encoding/json"
6+ "errors"
7+ "fmt"
8+ "image"
9+ _ "image/png"
10+ "os"
11+)
12+
13+const fixtureMargin = 100
14+const minimumChangedPixels = 20
15+
16+func load(path string) (image.Image, error) {
17+ f, err := os.Open(path)
18+ if err != nil {
19+ return nil, err
20+ }
21+ defer f.Close()
22+ im, _, err := image.Decode(f)
23+ return im, err
24+}
25+
26+func compare(a, b image.Image) (int, error) {
27+ if a.Bounds() != b.Bounds() {
28+ return 0, errors.New("capture dimensions changed")
29+ }
30+ bounds := a.Bounds()
31+ if bounds.Dx() <= 2*fixtureMargin || bounds.Dy() <= 2*fixtureMargin {
32+ return 0, errors.New("capture has no comparison region")
33+ }
34+ changed := 0
35+ for y := bounds.Min.Y + fixtureMargin; y < bounds.Max.Y-fixtureMargin; y++ {
36+ for x := bounds.Min.X + fixtureMargin; x < bounds.Max.X-fixtureMargin; x++ {
37+ ar, ag, ab, _ := a.At(x, y).RGBA()
38+ br, bg, bb, _ := b.At(x, y).RGBA()
39+ if ar != br || ag != bg || ab != bb {
40+ changed++
41+ }
42+ }
43+ }
44+ return changed, nil
45+}
46+
47+func execute(args []string) (int, error) {
48+ if len(args) != 2 {
49+ return 2, errors.New("usage: pixels before.png after.png")
50+ }
51+ a, err := load(args[0])
52+ if err != nil {
53+ return 2, err
54+ }
55+ b, err := load(args[1])
56+ if err != nil {
57+ return 2, err
58+ }
59+ changed, err := compare(a, b)
60+ if err != nil {
61+ return 2, err
62+ }
63+ if err := json.NewEncoder(os.Stdout).Encode(map[string]int{"changed_pixels": changed}); err != nil {
64+ return 2, err
65+ }
66+ if changed < minimumChangedPixels {
67+ return 1, nil
68+ }
69+ return 0, nil
70+}
71+func main() {
72+ code, err := execute(os.Args[1:])
73+ if err != nil {
74+ fmt.Fprintln(os.Stderr, err)
75+ }
76+ os.Exit(code)
77+}
A scripts/remote-resilience-harness/pixels/main_test.go +28 -0
@@ -0,0 +1,28 @@
1+package main
2+
3+import (
4+ "image"
5+ "image/color"
6+ "testing"
7+
8+ "github.com/stretchr/testify/require"
9+)
10+
11+func TestCompare(t *testing.T) {
12+ for _, n := range []int{0, 19, 20} {
13+ a := image.NewRGBA(image.Rect(0, 0, 300, 300))
14+ b := image.NewRGBA(a.Bounds())
15+ for x := 0; x < n; x++ {
16+ b.Set(100+x, 100, color.White)
17+ }
18+ got, err := compare(a, b)
19+ require.NoError(t, err)
20+ require.Equal(t, n, got)
21+ }
22+ for _, bounds := range []image.Rectangle{image.Rect(0, 0, 200, 300), image.Rect(0, 0, 300, 200)} {
23+ _, err := compare(image.NewRGBA(bounds), image.NewRGBA(bounds))
24+ require.Error(t, err)
25+ }
26+ _, err := compare(image.NewRGBA(image.Rect(0, 0, 300, 300)), image.NewRGBA(image.Rect(0, 0, 400, 300)))
27+ require.Error(t, err)
28+}
@@ -0,0 +1,28 @@
1+package main
2+
3+import (
4+ "image"
5+ "image/color"
6+ "testing"
7+
8+ "github.com/stretchr/testify/require"
9+)
10+
11+func TestCompare(t *testing.T) {
12+ for _, n := range []int{0, 19, 20} {
13+ a := image.NewRGBA(image.Rect(0, 0, 300, 300))
14+ b := image.NewRGBA(a.Bounds())
15+ for x := 0; x < n; x++ {
16+ b.Set(100+x, 100, color.White)
17+ }
18+ got, err := compare(a, b)
19+ require.NoError(t, err)
20+ require.Equal(t, n, got)
21+ }
22+ for _, bounds := range []image.Rectangle{image.Rect(0, 0, 200, 300), image.Rect(0, 0, 300, 200)} {
23+ _, err := compare(image.NewRGBA(bounds), image.NewRGBA(bounds))
24+ require.Error(t, err)
25+ }
26+ _, err := compare(image.NewRGBA(image.Rect(0, 0, 300, 300)), image.NewRGBA(image.Rect(0, 0, 400, 300)))
27+ require.Error(t, err)
28+}
A scripts/remote-resilience-harness/relay/main.go +289 -0
@@ -0,0 +1,289 @@
1+// Command relay is disposable acceptance infrastructure, never a production proxy.
2+package main
3+
4+import (
5+ "bufio"
6+ "bytes"
7+ "context"
8+ "encoding/json"
9+ "errors"
10+ "flag"
11+ "fmt"
12+ "io"
13+ "net"
14+ "os"
15+ "os/signal"
16+ "path/filepath"
17+ "sync"
18+ "sync/atomic"
19+ "syscall"
20+ "time"
21+
22+ "github.com/bnema/vev/internal/adapters/quicnettest"
23+)
24+
25+type command struct {
26+ ID uint64 `json:"id"`
27+ Op string `json:"op"`
28+ Port int `json:"port,omitempty"`
29+ UDP bool `json:"udp,omitempty"`
30+ TCP bool `json:"tcp,omitempty"`
31+ Active bool `json:"active,omitempty"`
32+}
33+
34+type relay struct {
35+ ctx context.Context
36+ cancel context.CancelFunc
37+ remote string
38+ degraded bool
39+ blackout atomic.Bool
40+ bytes atomic.Uint64
41+ mu sync.Mutex
42+ connections map[net.Conn]struct{}
43+ proxies []*quicnettest.Proxy
44+ udpBlackout bool
45+ wg sync.WaitGroup
46+}
47+
48+func main() {
49+ remote := flag.String("remote", "remote", "fixture upstream host")
50+ control := flag.String("control", "", "required private Unix control path")
51+ tcp := flag.String("tcp", ":2222", "SSH listener")
52+ degraded := flag.Bool("degraded", false, "enable seeded degraded profile")
53+ flag.Parse()
54+ if *control == "" {
55+ fmt.Fprintln(os.Stderr, "-control is required")
56+ os.Exit(2)
57+ }
58+ ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
59+ r := &relay{ctx: ctx, cancel: cancel, remote: *remote, degraded: *degraded, connections: make(map[net.Conn]struct{})}
60+ if err := r.serve(*tcp, *control); err != nil {
61+ fmt.Fprintln(os.Stderr, err)
62+ os.Exit(1)
63+ }
64+}
65+
66+func (r *relay) serve(addr, path string) error {
67+ listener, err := net.Listen("tcp", addr)
68+ if err != nil {
69+ return err
70+ }
71+ defer listener.Close()
72+ parent, err := os.Stat(filepath.Dir(path))
73+ if err != nil {
74+ return err
75+ }
76+ if !parent.IsDir() || parent.Mode().Perm() != 0700 {
77+ return errors.New("control socket requires a private 0700 directory")
78+ }
79+ owner, ok := parent.Sys().(*syscall.Stat_t)
80+ if !ok || owner.Uid != uint32(os.Getuid()) {
81+ return errors.New("control directory must belong to the current user")
82+ }
83+ control, err := net.Listen("unix", path)
84+ if err != nil {
85+ return err
86+ }
87+ defer control.Close()
88+ watchDone := make(chan struct{})
89+ defer close(watchDone)
90+ go func() {
91+ select {
92+ case <-r.ctx.Done():
93+ listener.Close()
94+ control.Close()
95+ case <-watchDone:
96+ }
97+ }()
98+ defer os.Remove(path)
99+ if err := os.Chmod(path, 0600); err != nil {
100+ return err
101+ }
102+ defer func() { listener.Close(); r.close() }()
103+ r.wg.Add(1)
104+ go func() {
105+ defer r.wg.Done()
106+ for {
107+ conn, err := listener.Accept()
108+ if err != nil {
109+ return
110+ }
111+ r.mu.Lock()
112+ if len(r.connections) >= 64 {
113+ r.mu.Unlock()
114+ conn.Close()
115+ continue
116+ }
117+ r.connections[conn] = struct{}{}
118+ r.mu.Unlock()
119+ r.wg.Add(1)
120+ go func() { defer r.wg.Done(); r.forward(conn) }()
121+ }
122+ }()
123+ for {
124+ conn, err := control.Accept()
125+ if err != nil {
126+ if r.ctx.Err() != nil {
127+ return nil
128+ }
129+ return err
130+ }
131+ // Commands are bounded and handled serially; a stalled caller gets 2s.
132+ conn.SetDeadline(time.Now().Add(2 * time.Second))
133+ scanner := bufio.NewScanner(conn)
134+ scanner.Buffer(make([]byte, 4096), 4096)
135+ if scanner.Scan() {
136+ var c command
137+ err := json.Unmarshal(scanner.Bytes(), &c)
138+ var result any
139+ if err == nil {
140+ result, err = r.apply(c)
141+ }
142+ reply := map[string]any{"id": c.ID, "result": result}
143+ if err != nil {
144+ reply["error"] = err.Error()
145+ }
146+ json.NewEncoder(conn).Encode(reply)
147+ }
148+ conn.Close()
149+ if r.ctx.Err() != nil {
150+ listener.Close()
151+ return nil
152+ }
153+ }
154+}
155+
156+func (r *relay) apply(c command) (any, error) {
157+ switch c.Op {
158+ case "udp":
159+ if c.Port <= 0 || c.Port > 65535 {
160+ return nil, errors.New("invalid port")
161+ }
162+ upstream, err := net.ResolveUDPAddr("udp", net.JoinHostPort(r.remote, fmt.Sprint(c.Port)))
163+ if err != nil {
164+ return nil, err
165+ }
166+ link := quicnettest.LinkConfig{}
167+ if r.degraded {
168+ link = quicnettest.LinkConfig{BaseLatency: 150 * time.Millisecond, Jitter: 50 * time.Millisecond, LossPercent: 5}
169+ }
170+ r.mu.Lock()
171+ defer r.mu.Unlock()
172+ if len(r.proxies) >= 32 {
173+ return nil, errors.New("UDP listener limit")
174+ }
175+ p, err := quicnettest.New(quicnettest.Config{ServerAddr: upstream, ListenAddr: &net.UDPAddr{IP: net.IPv4zero}, Seed: 317, ToServer: link, ToClient: link})
176+ if err != nil {
177+ return nil, err
178+ }
179+ p.SetBlackout(r.udpBlackout)
180+ r.proxies = append(r.proxies, p)
181+ return p.Addr().Port, nil
182+ case "blackout":
183+ if c.TCP {
184+ r.blackout.Store(c.Active)
185+ }
186+ if c.UDP {
187+ r.mu.Lock()
188+ r.udpBlackout = c.Active
189+ for _, p := range r.proxies {
190+ p.SetBlackout(c.Active)
191+ }
192+ r.mu.Unlock()
193+ }
194+ return nil, nil
195+ case "disconnect":
196+ r.mu.Lock()
197+ for conn := range r.connections {
198+ conn.Close()
199+ }
200+ r.mu.Unlock()
201+ return nil, nil
202+ case "stats":
203+ r.mu.Lock()
204+ defer r.mu.Unlock()
205+ stats := make([]quicnettest.Stats, 0, len(r.proxies))
206+ for _, p := range r.proxies {
207+ stats = append(stats, p.Stats())
208+ }
209+ return map[string]any{"udp": stats, "tcp_bytes": r.bytes.Load()}, nil
210+ case "stop":
211+ r.cancel()
212+ return nil, nil
213+ default:
214+ return nil, errors.New("unknown command")
215+ }
216+}
217+
218+func (r *relay) forward(client net.Conn) {
219+ defer func() { client.Close(); r.mu.Lock(); delete(r.connections, client); r.mu.Unlock() }()
220+ server, err := (&net.Dialer{Timeout: 5 * time.Second}).DialContext(r.ctx, "tcp", net.JoinHostPort(r.remote, "22"))
221+ if err != nil {
222+ return
223+ }
224+ defer server.Close()
225+ r.mu.Lock()
226+ r.connections[server] = struct{}{}
227+ r.mu.Unlock()
228+ defer func() { r.mu.Lock(); delete(r.connections, server); r.mu.Unlock() }()
229+ done := make(chan struct{})
230+ go func() { r.copy(server, client); server.Close(); client.Close(); close(done) }()
231+ r.copy(client, server)
232+ server.Close()
233+ client.Close()
234+ <-done
235+}
236+
237+func (r *relay) wait(delay time.Duration) bool {
238+ timer := time.NewTimer(delay)
239+ defer timer.Stop()
240+ select {
241+ case <-timer.C:
242+ return true
243+ case <-r.ctx.Done():
244+ return false
245+ }
246+}
247+
248+func (r *relay) copy(dst, src net.Conn) {
249+ buf := make([]byte, 32<<10)
250+ for {
251+ for r.blackout.Load() {
252+ if !r.wait(10 * time.Millisecond) {
253+ return
254+ }
255+ }
256+ n, err := src.Read(buf)
257+ if n > 0 {
258+ for r.blackout.Load() {
259+ if !r.wait(10 * time.Millisecond) {
260+ return
261+ }
262+ }
263+ if r.degraded && !r.wait(150*time.Millisecond+time.Duration(n)*time.Second/(64<<10)) {
264+ return
265+ }
266+ written, writeErr := io.Copy(dst, bytes.NewReader(buf[:n]))
267+ r.bytes.Add(uint64(written))
268+ if writeErr != nil {
269+ return
270+ }
271+ }
272+ if err != nil {
273+ return
274+ }
275+ }
276+}
277+
278+func (r *relay) close() {
279+ r.cancel()
280+ r.mu.Lock()
281+ for conn := range r.connections {
282+ conn.Close()
283+ }
284+ for _, p := range r.proxies {
285+ p.Close()
286+ }
287+ r.mu.Unlock()
288+ r.wg.Wait()
289+}
@@ -0,0 +1,289 @@
1+// Command relay is disposable acceptance infrastructure, never a production proxy.
2+package main
3+
4+import (
5+ "bufio"
6+ "bytes"
7+ "context"
8+ "encoding/json"
9+ "errors"
10+ "flag"
11+ "fmt"
12+ "io"
13+ "net"
14+ "os"
15+ "os/signal"
16+ "path/filepath"
17+ "sync"
18+ "sync/atomic"
19+ "syscall"
20+ "time"
21+
22+ "github.com/bnema/vev/internal/adapters/quicnettest"
23+)
24+
25+type command struct {
26+ ID uint64 `json:"id"`
27+ Op string `json:"op"`
28+ Port int `json:"port,omitempty"`
29+ UDP bool `json:"udp,omitempty"`
30+ TCP bool `json:"tcp,omitempty"`
31+ Active bool `json:"active,omitempty"`
32+}
33+
34+type relay struct {
35+ ctx context.Context
36+ cancel context.CancelFunc
37+ remote string
38+ degraded bool
39+ blackout atomic.Bool
40+ bytes atomic.Uint64
41+ mu sync.Mutex
42+ connections map[net.Conn]struct{}
43+ proxies []*quicnettest.Proxy
44+ udpBlackout bool
45+ wg sync.WaitGroup
46+}
47+
48+func main() {
49+ remote := flag.String("remote", "remote", "fixture upstream host")
50+ control := flag.String("control", "", "required private Unix control path")
51+ tcp := flag.String("tcp", ":2222", "SSH listener")
52+ degraded := flag.Bool("degraded", false, "enable seeded degraded profile")
53+ flag.Parse()
54+ if *control == "" {
55+ fmt.Fprintln(os.Stderr, "-control is required")
56+ os.Exit(2)
57+ }
58+ ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
59+ r := &relay{ctx: ctx, cancel: cancel, remote: *remote, degraded: *degraded, connections: make(map[net.Conn]struct{})}
60+ if err := r.serve(*tcp, *control); err != nil {
61+ fmt.Fprintln(os.Stderr, err)
62+ os.Exit(1)
63+ }
64+}
65+
66+func (r *relay) serve(addr, path string) error {
67+ listener, err := net.Listen("tcp", addr)
68+ if err != nil {
69+ return err
70+ }
71+ defer listener.Close()
72+ parent, err := os.Stat(filepath.Dir(path))
73+ if err != nil {
74+ return err
75+ }
76+ if !parent.IsDir() || parent.Mode().Perm() != 0700 {
77+ return errors.New("control socket requires a private 0700 directory")
78+ }
79+ owner, ok := parent.Sys().(*syscall.Stat_t)
80+ if !ok || owner.Uid != uint32(os.Getuid()) {
81+ return errors.New("control directory must belong to the current user")
82+ }
83+ control, err := net.Listen("unix", path)
84+ if err != nil {
85+ return err
86+ }
87+ defer control.Close()
88+ watchDone := make(chan struct{})
89+ defer close(watchDone)
90+ go func() {
91+ select {
92+ case <-r.ctx.Done():
93+ listener.Close()
94+ control.Close()
95+ case <-watchDone:
96+ }
97+ }()
98+ defer os.Remove(path)
99+ if err := os.Chmod(path, 0600); err != nil {
100+ return err
101+ }
102+ defer func() { listener.Close(); r.close() }()
103+ r.wg.Add(1)
104+ go func() {
105+ defer r.wg.Done()
106+ for {
107+ conn, err := listener.Accept()
108+ if err != nil {
109+ return
110+ }
111+ r.mu.Lock()
112+ if len(r.connections) >= 64 {
113+ r.mu.Unlock()
114+ conn.Close()
115+ continue
116+ }
117+ r.connections[conn] = struct{}{}
118+ r.mu.Unlock()
119+ r.wg.Add(1)
120+ go func() { defer r.wg.Done(); r.forward(conn) }()
121+ }
122+ }()
123+ for {
124+ conn, err := control.Accept()
125+ if err != nil {
126+ if r.ctx.Err() != nil {
127+ return nil
128+ }
129+ return err
130+ }
131+ // Commands are bounded and handled serially; a stalled caller gets 2s.
132+ conn.SetDeadline(time.Now().Add(2 * time.Second))
133+ scanner := bufio.NewScanner(conn)
134+ scanner.Buffer(make([]byte, 4096), 4096)
135+ if scanner.Scan() {
136+ var c command
137+ err := json.Unmarshal(scanner.Bytes(), &c)
138+ var result any
139+ if err == nil {
140+ result, err = r.apply(c)
141+ }
142+ reply := map[string]any{"id": c.ID, "result": result}
143+ if err != nil {
144+ reply["error"] = err.Error()
145+ }
146+ json.NewEncoder(conn).Encode(reply)
147+ }
148+ conn.Close()
149+ if r.ctx.Err() != nil {
150+ listener.Close()
151+ return nil
152+ }
153+ }
154+}
155+
156+func (r *relay) apply(c command) (any, error) {
157+ switch c.Op {
158+ case "udp":
159+ if c.Port <= 0 || c.Port > 65535 {
160+ return nil, errors.New("invalid port")
161+ }
162+ upstream, err := net.ResolveUDPAddr("udp", net.JoinHostPort(r.remote, fmt.Sprint(c.Port)))
163+ if err != nil {
164+ return nil, err
165+ }
166+ link := quicnettest.LinkConfig{}
167+ if r.degraded {
168+ link = quicnettest.LinkConfig{BaseLatency: 150 * time.Millisecond, Jitter: 50 * time.Millisecond, LossPercent: 5}
169+ }
170+ r.mu.Lock()
171+ defer r.mu.Unlock()
172+ if len(r.proxies) >= 32 {
173+ return nil, errors.New("UDP listener limit")
174+ }
175+ p, err := quicnettest.New(quicnettest.Config{ServerAddr: upstream, ListenAddr: &net.UDPAddr{IP: net.IPv4zero}, Seed: 317, ToServer: link, ToClient: link})
176+ if err != nil {
177+ return nil, err
178+ }
179+ p.SetBlackout(r.udpBlackout)
180+ r.proxies = append(r.proxies, p)
181+ return p.Addr().Port, nil
182+ case "blackout":
183+ if c.TCP {
184+ r.blackout.Store(c.Active)
185+ }
186+ if c.UDP {
187+ r.mu.Lock()
188+ r.udpBlackout = c.Active
189+ for _, p := range r.proxies {
190+ p.SetBlackout(c.Active)
191+ }
192+ r.mu.Unlock()
193+ }
194+ return nil, nil
195+ case "disconnect":
196+ r.mu.Lock()
197+ for conn := range r.connections {
198+ conn.Close()
199+ }
200+ r.mu.Unlock()
201+ return nil, nil
202+ case "stats":
203+ r.mu.Lock()
204+ defer r.mu.Unlock()
205+ stats := make([]quicnettest.Stats, 0, len(r.proxies))
206+ for _, p := range r.proxies {
207+ stats = append(stats, p.Stats())
208+ }
209+ return map[string]any{"udp": stats, "tcp_bytes": r.bytes.Load()}, nil
210+ case "stop":
211+ r.cancel()
212+ return nil, nil
213+ default:
214+ return nil, errors.New("unknown command")
215+ }
216+}
217+
218+func (r *relay) forward(client net.Conn) {
219+ defer func() { client.Close(); r.mu.Lock(); delete(r.connections, client); r.mu.Unlock() }()
220+ server, err := (&net.Dialer{Timeout: 5 * time.Second}).DialContext(r.ctx, "tcp", net.JoinHostPort(r.remote, "22"))
221+ if err != nil {
222+ return
223+ }
224+ defer server.Close()
225+ r.mu.Lock()
226+ r.connections[server] = struct{}{}
227+ r.mu.Unlock()
228+ defer func() { r.mu.Lock(); delete(r.connections, server); r.mu.Unlock() }()
229+ done := make(chan struct{})
230+ go func() { r.copy(server, client); server.Close(); client.Close(); close(done) }()
231+ r.copy(client, server)
232+ server.Close()
233+ client.Close()
234+ <-done
235+}
236+
237+func (r *relay) wait(delay time.Duration) bool {
238+ timer := time.NewTimer(delay)
239+ defer timer.Stop()
240+ select {
241+ case <-timer.C:
242+ return true
243+ case <-r.ctx.Done():
244+ return false
245+ }
246+}
247+
248+func (r *relay) copy(dst, src net.Conn) {
249+ buf := make([]byte, 32<<10)
250+ for {
251+ for r.blackout.Load() {
252+ if !r.wait(10 * time.Millisecond) {
253+ return
254+ }
255+ }
256+ n, err := src.Read(buf)
257+ if n > 0 {
258+ for r.blackout.Load() {
259+ if !r.wait(10 * time.Millisecond) {
260+ return
261+ }
262+ }
263+ if r.degraded && !r.wait(150*time.Millisecond+time.Duration(n)*time.Second/(64<<10)) {
264+ return
265+ }
266+ written, writeErr := io.Copy(dst, bytes.NewReader(buf[:n]))
267+ r.bytes.Add(uint64(written))
268+ if writeErr != nil {
269+ return
270+ }
271+ }
272+ if err != nil {
273+ return
274+ }
275+ }
276+}
277+
278+func (r *relay) close() {
279+ r.cancel()
280+ r.mu.Lock()
281+ for conn := range r.connections {
282+ conn.Close()
283+ }
284+ for _, p := range r.proxies {
285+ p.Close()
286+ }
287+ r.mu.Unlock()
288+ r.wg.Wait()
289+}
A scripts/remote-resilience-harness/relay/main_test.go +162 -0
@@ -0,0 +1,162 @@
1+package main
2+
3+import (
4+ "bufio"
5+ "context"
6+ "encoding/json"
7+ "io"
8+ "net"
9+ "os"
10+ "path/filepath"
11+ "testing"
12+ "time"
13+
14+ "github.com/stretchr/testify/require"
15+)
16+
17+func TestTCPRelayPreservesBytesAndStops(t *testing.T) {
18+ ctx, cancel := context.WithCancel(t.Context())
19+ defer cancel()
20+ r := &relay{ctx: ctx, cancel: cancel}
21+ source, writer := net.Pipe()
22+ reader, destination := net.Pipe()
23+ defer source.Close()
24+ defer writer.Close()
25+ defer reader.Close()
26+ defer destination.Close()
27+ done := make(chan struct{})
28+ go func() { defer close(done); r.copy(destination, source) }()
29+ payload := []byte("fixture input in order")
30+ go func() { writer.Write(payload); writer.Close() }()
31+ reader.SetReadDeadline(time.Now().Add(time.Second))
32+ got := make([]byte, len(payload))
33+ _, err := io.ReadFull(reader, got)
34+ require.NoError(t, err)
35+ require.Equal(t, payload, got)
36+ select {
37+ case <-done:
38+ case <-time.After(time.Second):
39+ t.Fatal("relay did not stop after EOF")
40+ }
41+ require.Equal(t, uint64(len(payload)), r.bytes.Load())
42+}
43+
44+func TestRelayShutdownClosesListeners(t *testing.T) {
45+ ctx, cancel := context.WithCancel(t.Context())
46+ r := &relay{ctx: ctx, cancel: cancel, connections: make(map[net.Conn]struct{})}
47+ directory := t.TempDir()
48+ require.NoError(t, os.Chmod(directory, 0700))
49+ path := filepath.Join(directory, "control.sock")
50+ done := make(chan error, 1)
51+ go func() { done <- r.serve("127.0.0.1:0", path) }()
52+ require.Eventually(t, func() bool { _, err := os.Stat(path); return err == nil }, time.Second, time.Millisecond)
53+ cancel()
54+ select {
55+ case err := <-done:
56+ require.NoError(t, err)
57+ case <-time.After(time.Second):
58+ t.Fatal("relay shutdown hung")
59+ }
60+ _, err := os.Stat(path)
61+ require.True(t, os.IsNotExist(err))
62+}
63+
64+func TestTCPRelayBlackoutHoldsBytes(t *testing.T) {
65+ ctx, cancel := context.WithCancel(t.Context())
66+ defer cancel()
67+ r := &relay{ctx: ctx, cancel: cancel}
68+ _, err := r.apply(command{Op: "blackout", TCP: true, Active: true})
69+ require.NoError(t, err)
70+ source, writer := net.Pipe()
71+ reader, destination := net.Pipe()
72+ defer source.Close()
73+ defer writer.Close()
74+ defer reader.Close()
75+ defer destination.Close()
76+ done := make(chan struct{})
77+ go func() { defer close(done); r.copy(destination, source) }()
78+ payload := []byte("held")
79+ wrote := make(chan error, 1)
80+ go func() { _, err := writer.Write(payload); wrote <- err }()
81+ select {
82+ case err := <-wrote:
83+ t.Fatalf("blackout read the source: %v", err)
84+ case <-time.After(50 * time.Millisecond):
85+ }
86+ require.Zero(t, r.bytes.Load())
87+ _, err = r.apply(command{Op: "blackout", TCP: true, Active: false})
88+ require.NoError(t, err)
89+ reader.SetReadDeadline(time.Now().Add(time.Second))
90+ got := make([]byte, len(payload))
91+ _, err = io.ReadFull(reader, got)
92+ require.NoError(t, err)
93+ require.Equal(t, payload, got)
94+ require.NoError(t, <-wrote)
95+ cancel()
96+ writer.Close()
97+ <-done
98+}
99+
100+func TestRelayDisconnectClosesConnections(t *testing.T) {
101+ ctx, cancel := context.WithCancel(t.Context())
102+ defer cancel()
103+ r := &relay{ctx: ctx, cancel: cancel, connections: make(map[net.Conn]struct{})}
104+ local, peer := net.Pipe()
105+ defer peer.Close()
106+ r.connections[local] = struct{}{}
107+ _, err := r.apply(command{Op: "disconnect"})
108+ require.NoError(t, err)
109+ peer.SetReadDeadline(time.Now().Add(time.Second))
110+ _, err = peer.Read(make([]byte, 1))
111+ require.ErrorIs(t, err, io.EOF)
112+}
113+
114+func TestRelayControlRoundTrip(t *testing.T) {
115+ ctx, cancel := context.WithCancel(t.Context())
116+ defer cancel()
117+ r := &relay{ctx: ctx, cancel: cancel, connections: make(map[net.Conn]struct{})}
118+ directory := t.TempDir()
119+ require.NoError(t, os.Chmod(directory, 0700))
120+ path := filepath.Join(directory, "control.sock")
121+ done := make(chan error, 1)
122+ go func() { done <- r.serve("127.0.0.1:0", path) }()
123+ require.Eventually(t, func() bool { _, err := os.Stat(path); return err == nil }, time.Second, time.Millisecond)
124+ send := func(c command) map[string]any {
125+ conn, err := net.Dial("unix", path)
126+ require.NoError(t, err)
127+ defer conn.Close()
128+ require.NoError(t, json.NewEncoder(conn).Encode(c))
129+ line, err := bufio.NewReader(conn).ReadBytes('\n')
130+ require.NoError(t, err)
131+ var reply map[string]any
132+ require.NoError(t, json.Unmarshal(line, &reply))
133+ return reply
134+ }
135+ reply := send(command{ID: 7, Op: "blackout", TCP: true, Active: true})
136+ require.EqualValues(t, 7, reply["id"])
137+ require.NotContains(t, reply, "error")
138+ require.True(t, r.blackout.Load())
139+ reply = send(command{ID: 8, Op: "unknown"})
140+ require.Equal(t, "unknown command", reply["error"])
141+ reply = send(command{ID: 9, Op: "stop"})
142+ require.NotContains(t, reply, "error")
143+ select {
144+ case err := <-done:
145+ require.NoError(t, err)
146+ case <-time.After(time.Second):
147+ t.Fatal("stop command did not shut the relay down")
148+ }
149+}
150+
151+func TestRelayRejectsInvalidControl(t *testing.T) {
152+ ctx, cancel := context.WithCancel(t.Context())
153+ defer cancel()
154+ r := &relay{ctx: ctx, cancel: cancel}
155+ for _, c := range []command{{Op: "unknown"}, {Op: "udp", Port: 0}, {Op: "udp", Port: 65536}} {
156+ _, err := r.apply(c)
157+ require.Error(t, err)
158+ }
159+ _, err := r.apply(command{Op: "stop"})
160+ require.NoError(t, err)
161+ require.ErrorIs(t, ctx.Err(), context.Canceled)
162+}
@@ -0,0 +1,162 @@
1+package main
2+
3+import (
4+ "bufio"
5+ "context"
6+ "encoding/json"
7+ "io"
8+ "net"
9+ "os"
10+ "path/filepath"
11+ "testing"
12+ "time"
13+
14+ "github.com/stretchr/testify/require"
15+)
16+
17+func TestTCPRelayPreservesBytesAndStops(t *testing.T) {
18+ ctx, cancel := context.WithCancel(t.Context())
19+ defer cancel()
20+ r := &relay{ctx: ctx, cancel: cancel}
21+ source, writer := net.Pipe()
22+ reader, destination := net.Pipe()
23+ defer source.Close()
24+ defer writer.Close()
25+ defer reader.Close()
26+ defer destination.Close()
27+ done := make(chan struct{})
28+ go func() { defer close(done); r.copy(destination, source) }()
29+ payload := []byte("fixture input in order")
30+ go func() { writer.Write(payload); writer.Close() }()
31+ reader.SetReadDeadline(time.Now().Add(time.Second))
32+ got := make([]byte, len(payload))
33+ _, err := io.ReadFull(reader, got)
34+ require.NoError(t, err)
35+ require.Equal(t, payload, got)
36+ select {
37+ case <-done:
38+ case <-time.After(time.Second):
39+ t.Fatal("relay did not stop after EOF")
40+ }
41+ require.Equal(t, uint64(len(payload)), r.bytes.Load())
42+}
43+
44+func TestRelayShutdownClosesListeners(t *testing.T) {
45+ ctx, cancel := context.WithCancel(t.Context())
46+ r := &relay{ctx: ctx, cancel: cancel, connections: make(map[net.Conn]struct{})}
47+ directory := t.TempDir()
48+ require.NoError(t, os.Chmod(directory, 0700))
49+ path := filepath.Join(directory, "control.sock")
50+ done := make(chan error, 1)
51+ go func() { done <- r.serve("127.0.0.1:0", path) }()
52+ require.Eventually(t, func() bool { _, err := os.Stat(path); return err == nil }, time.Second, time.Millisecond)
53+ cancel()
54+ select {
55+ case err := <-done:
56+ require.NoError(t, err)
57+ case <-time.After(time.Second):
58+ t.Fatal("relay shutdown hung")
59+ }
60+ _, err := os.Stat(path)
61+ require.True(t, os.IsNotExist(err))
62+}
63+
64+func TestTCPRelayBlackoutHoldsBytes(t *testing.T) {
65+ ctx, cancel := context.WithCancel(t.Context())
66+ defer cancel()
67+ r := &relay{ctx: ctx, cancel: cancel}
68+ _, err := r.apply(command{Op: "blackout", TCP: true, Active: true})
69+ require.NoError(t, err)
70+ source, writer := net.Pipe()
71+ reader, destination := net.Pipe()
72+ defer source.Close()
73+ defer writer.Close()
74+ defer reader.Close()
75+ defer destination.Close()
76+ done := make(chan struct{})
77+ go func() { defer close(done); r.copy(destination, source) }()
78+ payload := []byte("held")
79+ wrote := make(chan error, 1)
80+ go func() { _, err := writer.Write(payload); wrote <- err }()
81+ select {
82+ case err := <-wrote:
83+ t.Fatalf("blackout read the source: %v", err)
84+ case <-time.After(50 * time.Millisecond):
85+ }
86+ require.Zero(t, r.bytes.Load())
87+ _, err = r.apply(command{Op: "blackout", TCP: true, Active: false})
88+ require.NoError(t, err)
89+ reader.SetReadDeadline(time.Now().Add(time.Second))
90+ got := make([]byte, len(payload))
91+ _, err = io.ReadFull(reader, got)
92+ require.NoError(t, err)
93+ require.Equal(t, payload, got)
94+ require.NoError(t, <-wrote)
95+ cancel()
96+ writer.Close()
97+ <-done
98+}
99+
100+func TestRelayDisconnectClosesConnections(t *testing.T) {
101+ ctx, cancel := context.WithCancel(t.Context())
102+ defer cancel()
103+ r := &relay{ctx: ctx, cancel: cancel, connections: make(map[net.Conn]struct{})}
104+ local, peer := net.Pipe()
105+ defer peer.Close()
106+ r.connections[local] = struct{}{}
107+ _, err := r.apply(command{Op: "disconnect"})
108+ require.NoError(t, err)
109+ peer.SetReadDeadline(time.Now().Add(time.Second))
110+ _, err = peer.Read(make([]byte, 1))
111+ require.ErrorIs(t, err, io.EOF)
112+}
113+
114+func TestRelayControlRoundTrip(t *testing.T) {
115+ ctx, cancel := context.WithCancel(t.Context())
116+ defer cancel()
117+ r := &relay{ctx: ctx, cancel: cancel, connections: make(map[net.Conn]struct{})}
118+ directory := t.TempDir()
119+ require.NoError(t, os.Chmod(directory, 0700))
120+ path := filepath.Join(directory, "control.sock")
121+ done := make(chan error, 1)
122+ go func() { done <- r.serve("127.0.0.1:0", path) }()
123+ require.Eventually(t, func() bool { _, err := os.Stat(path); return err == nil }, time.Second, time.Millisecond)
124+ send := func(c command) map[string]any {
125+ conn, err := net.Dial("unix", path)
126+ require.NoError(t, err)
127+ defer conn.Close()
128+ require.NoError(t, json.NewEncoder(conn).Encode(c))
129+ line, err := bufio.NewReader(conn).ReadBytes('\n')
130+ require.NoError(t, err)
131+ var reply map[string]any
132+ require.NoError(t, json.Unmarshal(line, &reply))
133+ return reply
134+ }
135+ reply := send(command{ID: 7, Op: "blackout", TCP: true, Active: true})
136+ require.EqualValues(t, 7, reply["id"])
137+ require.NotContains(t, reply, "error")
138+ require.True(t, r.blackout.Load())
139+ reply = send(command{ID: 8, Op: "unknown"})
140+ require.Equal(t, "unknown command", reply["error"])
141+ reply = send(command{ID: 9, Op: "stop"})
142+ require.NotContains(t, reply, "error")
143+ select {
144+ case err := <-done:
145+ require.NoError(t, err)
146+ case <-time.After(time.Second):
147+ t.Fatal("stop command did not shut the relay down")
148+ }
149+}
150+
151+func TestRelayRejectsInvalidControl(t *testing.T) {
152+ ctx, cancel := context.WithCancel(t.Context())
153+ defer cancel()
154+ r := &relay{ctx: ctx, cancel: cancel}
155+ for _, c := range []command{{Op: "unknown"}, {Op: "udp", Port: 0}, {Op: "udp", Port: 65536}} {
156+ _, err := r.apply(c)
157+ require.Error(t, err)
158+ }
159+ _, err := r.apply(command{Op: "stop"})
160+ require.NoError(t, err)
161+ require.ErrorIs(t, ctx.Err(), context.Canceled)
162+}
A scripts/remote-resilience-harness/ssh.py +58 -0
@@ -0,0 +1,58 @@
1+#!/usr/bin/env python3
2+"""Fixture-only SSH wrapper: pin SSH and rewrite only QUIC readiness port."""
3+import json
4+import os
5+import socket
6+import subprocess
7+import sys
8+
9+
10+def control(request):
11+ with socket.socket(socket.AF_UNIX) as conn:
12+ conn.settimeout(2)
13+ conn.connect(os.environ.get("RESILIENCE_CONTROL", "/home/demo/relay/control.sock"))
14+ conn.sendall(json.dumps(request).encode() + b"\n")
15+ reply = json.loads(conn.makefile("rb").readline(4096))
16+ if "error" in reply:
17+ raise RuntimeError(reply["error"])
18+ return reply["result"]
19+
20+
21+def main():
22+ args = sys.argv[1:]
23+ command = ["/usr/bin/ssh-real", "-p", "2222", "-o", "HostKeyAlias=remote"] + args
24+ if not any("_broker-mux-quic-bootstrap" in arg for arg in args):
25+ os.execv(command[0], command)
26+ process = subprocess.Popen(command, stdin=sys.stdin.buffer, stdout=subprocess.PIPE)
27+ try:
28+ line = process.stdout.readline(4097)
29+ if len(line) > 4096 or not line.endswith(b"\n"):
30+ raise RuntimeError("invalid bounded bootstrap readiness")
31+ readiness = json.loads(line)
32+ port = readiness.get("port")
33+ if not isinstance(port, int) or not 0 < port <= 65535:
34+ raise RuntimeError("invalid bootstrap port")
35+ readiness["port"] = control({"id": 1, "op": "udp", "port": port})
36+ sys.stdout.buffer.write(json.dumps(readiness, separators=(",", ":")).encode() + b"\n")
37+ sys.stdout.buffer.flush()
38+ # No readiness or encrypted payload is recorded in artifacts.
39+ while chunk := process.stdout.read1(32768):
40+ sys.stdout.buffer.write(chunk)
41+ sys.stdout.buffer.flush()
42+ return process.wait()
43+ finally:
44+ if process.poll() is None:
45+ process.terminate()
46+ try:
47+ process.wait(timeout=2)
48+ except subprocess.TimeoutExpired:
49+ process.kill()
50+ process.wait()
51+
52+
53+if __name__ == "__main__":
54+ try:
55+ sys.exit(main())
56+ except (OSError, ValueError, RuntimeError):
57+ print("resilience SSH fixture failed", file=sys.stderr)
58+ sys.exit(1)
@@ -0,0 +1,58 @@
1+#!/usr/bin/env python3
2+"""Fixture-only SSH wrapper: pin SSH and rewrite only QUIC readiness port."""
3+import json
4+import os
5+import socket
6+import subprocess
7+import sys
8+
9+
10+def control(request):
11+ with socket.socket(socket.AF_UNIX) as conn:
12+ conn.settimeout(2)
13+ conn.connect(os.environ.get("RESILIENCE_CONTROL", "/home/demo/relay/control.sock"))
14+ conn.sendall(json.dumps(request).encode() + b"\n")
15+ reply = json.loads(conn.makefile("rb").readline(4096))
16+ if "error" in reply:
17+ raise RuntimeError(reply["error"])
18+ return reply["result"]
19+
20+
21+def main():
22+ args = sys.argv[1:]
23+ command = ["/usr/bin/ssh-real", "-p", "2222", "-o", "HostKeyAlias=remote"] + args
24+ if not any("_broker-mux-quic-bootstrap" in arg for arg in args):
25+ os.execv(command[0], command)
26+ process = subprocess.Popen(command, stdin=sys.stdin.buffer, stdout=subprocess.PIPE)
27+ try:
28+ line = process.stdout.readline(4097)
29+ if len(line) > 4096 or not line.endswith(b"\n"):
30+ raise RuntimeError("invalid bounded bootstrap readiness")
31+ readiness = json.loads(line)
32+ port = readiness.get("port")
33+ if not isinstance(port, int) or not 0 < port <= 65535:
34+ raise RuntimeError("invalid bootstrap port")
35+ readiness["port"] = control({"id": 1, "op": "udp", "port": port})
36+ sys.stdout.buffer.write(json.dumps(readiness, separators=(",", ":")).encode() + b"\n")
37+ sys.stdout.buffer.flush()
38+ # No readiness or encrypted payload is recorded in artifacts.
39+ while chunk := process.stdout.read1(32768):
40+ sys.stdout.buffer.write(chunk)
41+ sys.stdout.buffer.flush()
42+ return process.wait()
43+ finally:
44+ if process.poll() is None:
45+ process.terminate()
46+ try:
47+ process.wait(timeout=2)
48+ except subprocess.TimeoutExpired:
49+ process.kill()
50+ process.wait()
51+
52+
53+if __name__ == "__main__":
54+ try:
55+ sys.exit(main())
56+ except (OSError, ValueError, RuntimeError):
57+ print("resilience SSH fixture failed", file=sys.stderr)
58+ sys.exit(1)
A scripts/remote-resilience-harness/ssh_test.py +30 -0
@@ -0,0 +1,30 @@
1+"""Fixture readiness relay tests; no network or production credentials."""
2+import importlib.util
3+from pathlib import Path
4+import unittest
5+from unittest.mock import patch, MagicMock
6+
7+spec = importlib.util.spec_from_file_location("fixture_ssh", Path(__file__).with_name("ssh.py"))
8+wrapper = importlib.util.module_from_spec(spec)
9+spec.loader.exec_module(wrapper)
10+
11+
12+class ControlTests(unittest.TestCase):
13+ def test_control_failure_is_not_a_port(self):
14+ conn = MagicMock()
15+ conn.__enter__.return_value = conn
16+ conn.makefile.return_value.readline.return_value = b'{"id":1,"error":"listener limit"}\n'
17+ with patch.object(wrapper.socket, "socket", return_value=conn):
18+ with self.assertRaises(RuntimeError):
19+ wrapper.control({"id": 1, "op": "udp", "port": 1234})
20+
21+ def test_control_returns_fixture_port(self):
22+ conn = MagicMock()
23+ conn.__enter__.return_value = conn
24+ conn.makefile.return_value.readline.return_value = b'{"id":1,"result":4321}\n'
25+ with patch.object(wrapper.socket, "socket", return_value=conn):
26+ self.assertEqual(wrapper.control({"id": 1, "op": "udp", "port": 1234}), 4321)
27+
28+
29+if __name__ == "__main__":
30+ unittest.main()
@@ -0,0 +1,30 @@
1+"""Fixture readiness relay tests; no network or production credentials."""
2+import importlib.util
3+from pathlib import Path
4+import unittest
5+from unittest.mock import patch, MagicMock
6+
7+spec = importlib.util.spec_from_file_location("fixture_ssh", Path(__file__).with_name("ssh.py"))
8+wrapper = importlib.util.module_from_spec(spec)
9+spec.loader.exec_module(wrapper)
10+
11+
12+class ControlTests(unittest.TestCase):
13+ def test_control_failure_is_not_a_port(self):
14+ conn = MagicMock()
15+ conn.__enter__.return_value = conn
16+ conn.makefile.return_value.readline.return_value = b'{"id":1,"error":"listener limit"}\n'
17+ with patch.object(wrapper.socket, "socket", return_value=conn):
18+ with self.assertRaises(RuntimeError):
19+ wrapper.control({"id": 1, "op": "udp", "port": 1234})
20+
21+ def test_control_returns_fixture_port(self):
22+ conn = MagicMock()
23+ conn.__enter__.return_value = conn
24+ conn.makefile.return_value.readline.return_value = b'{"id":1,"result":4321}\n'
25+ with patch.object(wrapper.socket, "socket", return_value=conn):
26+ self.assertEqual(wrapper.control({"id": 1, "op": "udp", "port": 1234}), 4321)
27+
28+
29+if __name__ == "__main__":
30+ unittest.main()
A scripts/remote-resilience-harness/visual.py +361 -0
@@ -0,0 +1,361 @@
1+#!/usr/bin/env python3
2+"""Disposable dual-transport fixture and physical Wayland input acceptance."""
3+import importlib.util
4+import json
5+import os
6+from pathlib import Path
7+import shutil
8+import signal
9+import subprocess
10+import tempfile
11+import time
12+import uuid
13+
14+ROOT = Path(__file__).resolve().parents[2]
15+
16+
17+def run(*args, **kwargs):
18+ return subprocess.run(args, check=True, timeout=kwargs.pop("timeout", 60), **kwargs)
19+
20+
21+def docker(*args, **kwargs):
22+ return run("docker", *args, **kwargs)
23+
24+
25+def main():
26+ name = "vev-resilience-" + uuid.uuid4().hex[:10]
27+ artifacts = Path(tempfile.mkdtemp(prefix="vev-resilience-artifacts-"))
28+ processes = []
29+ containers = []
30+ network = False
31+ def interrupt(signum, frame):
32+ raise KeyboardInterrupt("fixture interrupted")
33+ signal.signal(signal.SIGTERM, interrupt)
34+ with tempfile.TemporaryDirectory(prefix="vev-resilience-private-") as directory:
35+ private = Path(directory)
36+ runtime = private / "runtime"
37+ runtime.mkdir(mode=0o700)
38+ try:
39+ for tool in ("docker", "foot", "ssh-keygen", "go"):
40+ if shutil.which(tool) is None:
41+ raise RuntimeError("missing prerequisite: " + tool)
42+ run("go", "build", "-o", str(ROOT / "build/vev"), ".", cwd=ROOT)
43+ for helper in ("relay", "pixels"):
44+ run("go", "build", "-o", str(ROOT / f"build/resilience-{helper}"), f"./scripts/remote-resilience-harness/{helper}", cwd=ROOT)
45+ nefer = os.environ.get("NEFERWL_BIN")
46+ if not nefer:
47+ nefer = str(private / "neferwl")
48+ common = run("git", "rev-parse", "--path-format=absolute", "--git-common-dir", cwd=ROOT, capture_output=True, text=True).stdout.strip()
49+ source = Path(os.environ.get("NEFERWL_SOURCE", str(Path(common).parent.parent / "neferwl")))
50+ run("go", "build", "-o", nefer, "./cmd/neferwl", cwd=source, env={**os.environ, "CGO_ENABLED": "0"})
51+ docker("build", "-f", str(ROOT / "scripts/demo/Dockerfile"), "-t", name, str(ROOT), timeout=600)
52+ docker("network", "create", name)
53+ network = True
54+ for role in ("client", "remote"):
55+ container = name + "-" + role
56+ docker("run", "-d", "--name", container, "--network", name, "--network-alias", "upstream" if role == "remote" else "remote", "--entrypoint", "sleep", name, "infinity")
57+ containers.append(container)
58+ client, remote = containers
59+ for key in ("client", "host"):
60+ run("ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "fixture", "-f", str(private / key))
61+ docker("cp", str(private / "host"), remote + ":/etc/ssh/ssh_host_ed25519_key")
62+ docker("cp", str(private / "host.pub"), remote + ":/etc/ssh/ssh_host_ed25519_key.pub")
63+ docker("cp", str(private / "client.pub"), remote + ":/home/demo/.ssh/authorized_keys")
64+ docker("exec", "--user", "root", remote, "sh", "-c", "rm -f /home/demo/.ssh/id_ed25519*; chmod 600 /etc/ssh/ssh_host_ed25519_key /home/demo/.ssh/authorized_keys; chown demo:demo /home/demo/.ssh/authorized_keys; mkdir -p /run/sshd; /usr/sbin/sshd")
65+ docker("exec", "--user", "root", client, "sh", "-c", "rm -f /home/demo/.ssh/id_ed25519*")
66+ docker("cp", str(private / "client"), client + ":/home/demo/.ssh/id_ed25519")
67+ public = (private / "host.pub").read_text().split()
68+ (private / "known_hosts").write_text("remote " + " ".join(public[:2]) + "\n")
69+ (private / "config").write_text("Host remote\n HostName remote\n Port 2222\n User demo\n IdentityFile ~/.ssh/id_ed25519\n IdentitiesOnly yes\n StrictHostKeyChecking yes\n HostKeyAlias remote\n UserKnownHostsFile ~/.ssh/known_hosts\n")
70+ for file in ("known_hosts", "config"):
71+ docker("cp", str(private / file), client + ":/home/demo/.ssh/" + file)
72+ docker("cp", str(ROOT / "build/resilience-relay"), client + ":/tmp/relay")
73+ docker("cp", str(ROOT / "scripts/remote-resilience-harness/ssh.py"), client + ":/tmp/ssh")
74+ docker("exec", "--user", "root", client, "sh", "-c", "chown demo:demo /home/demo/.ssh/*; chmod 600 /home/demo/.ssh/*; chmod 755 /tmp/ssh; install -d -m 700 -o demo -g demo /home/demo/relay; mkdir -p /tmp/bin; ln -s /tmp/ssh /tmp/bin/ssh; mv /usr/bin/ssh /usr/bin/ssh-real; ln -s /tmp/ssh /usr/bin/ssh")
75+ transport = os.environ.get("VEV_RESILIENCE_TRANSPORT", "quic")
76+ if transport not in ("quic", "ssh"):
77+ raise ValueError("VEV_RESILIENCE_TRANSPORT must be quic or ssh")
78+ relay_args = ["-degraded"] if os.environ.get("VEV_RESILIENCE_DEGRADED") == "1" else []
79+ relay = subprocess.Popen(["docker", "exec", client, "/tmp/relay", "-remote", "upstream", "-control", "/home/demo/relay/control.sock", *relay_args])
80+ processes.append(relay)
81+ control_id = 0
82+ def control(op, **fields):
83+ nonlocal control_id
84+ control_id += 1
85+ request = {"id": control_id, "op": op, **fields}
86+ code = "import socket,json; s=socket.socket(socket.AF_UNIX); s.settimeout(3); s.connect('/home/demo/relay/control.sock'); s.sendall(" + repr(json.dumps(request).encode() + b"\n") + "); print(s.makefile().readline())"
87+ result = docker("exec", client, "python3", "-c", code, capture_output=True, text=True)
88+ reply = json.loads(result.stdout)
89+ assert reply["id"] == control_id and "error" not in reply, reply
90+ return reply["result"]
91+ # Fixture image must explicitly supply Python rather than using host tools.
92+ docker("exec", "--user", "root", client, "sh", "-c", "command -v python3", capture_output=True)
93+ deadline = time.monotonic() + 5
94+ while True:
95+ ready = subprocess.run(["docker", "exec", client, "test", "-S", "/home/demo/relay/control.sock"], capture_output=True, timeout=3)
96+ if ready.returncode == 0:
97+ break
98+ if time.monotonic() > deadline:
99+ raise RuntimeError("relay control socket did not become ready")
100+ time.sleep(0.05)
101+ probe = subprocess.run(["docker", "exec", client, "/usr/bin/ssh", "-v", "remote", "true"], capture_output=True, text=True, timeout=10)
102+ if probe.returncode:
103+ (artifacts / "ssh-probe.log").write_text(probe.stderr)
104+ raise RuntimeError("fixture SSH authentication failed")
105+ env_args = ["-e", "PATH=/tmp/bin:/usr/local/bin:/usr/bin", "-e", "VEV_LOG=debug", "-e", "SSH_AUTH_SOCK="]
106+ spec = importlib.util.spec_from_file_location("acceptance", ROOT / "scripts/remote-picker-harness/acceptance.py")
107+ acceptance = importlib.util.module_from_spec(spec)
108+ spec.loader.exec_module(acceptance)
109+ fixture = acceptance.Driver(remote, ["--session", "resilience"])
110+ fixture.close()
111+ docker("exec", *env_args, client, "vev", "host", "add", "--transport", transport, "remote")
112+ # Wait for the broker publication before resolving direct attach.
113+ deadline = time.monotonic() + 30
114+ while True:
115+ inventory = docker("exec", *env_args, client, "vev", "ls", "remote", capture_output=True, text=True)
116+ if "resilience" in inventory.stdout:
117+ break
118+ (artifacts / "inventory.txt").write_text(inventory.stdout + inventory.stderr)
119+ if time.monotonic() > deadline:
120+ raise RuntimeError("remote fixture was not published")
121+ time.sleep(0.2)
122+ shots = artifacts / "shots"
123+ shots.mkdir()
124+ docker_endpoint = run("docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}", capture_output=True, text=True).stdout.strip()
125+ env = {**os.environ, "DOCKER_HOST": docker_endpoint, "XDG_RUNTIME_DIR": str(runtime), "XDG_CONFIG_HOME": str(private / "config-home"), "XDG_STATE_HOME": str(private / "state"), "HOME": str(private)}
126+ env["TERMINAL"] = "foot --config=/dev/null --override=cursor.blink=no --log-level=info --hold -- docker --host=" + docker_endpoint + " exec -it " + " ".join(env_args) + " " + client + " vev attach demo@remote:resilience --ui-observe --ui-socket /home/demo/visual/ui.sock"
127+ log = (artifacts / "compositor.log").open("w")
128+ compositor = subprocess.Popen([nefer, "--backend=headless", "--size=1280x720", "--input=-", "--screenshot", str(shots), "--timeout", "180s"], stdin=subprocess.PIPE, stdout=log, stderr=log, text=True, env=env)
129+ processes.append(compositor)
130+ deadline = time.monotonic() + 15
131+ while not list(runtime.glob("wayland-*")):
132+ if compositor.poll() is not None or time.monotonic() > deadline:
133+ raise RuntimeError("headless compositor did not create socket")
134+ time.sleep(0.05)
135+ sockets = [p for p in runtime.glob("wayland-*") if not p.name.endswith(".lock")]
136+ env["WAYLAND_DISPLAY"] = str(sockets[0])
137+ env.pop("WAYLAND_SOCKET", None)
138+ env.pop("DISPLAY", None)
139+ def capture():
140+ bridge = subprocess.Popen(["docker", "exec", "-i", client, "vev", "--ui-driver", "--socket", "/home/demo/visual/ui.sock"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
141+ try:
142+ import select
143+ if not select.select([bridge.stdout], [], [], 1)[0]:
144+ raise RuntimeError("observation discovery timed out")
145+ discovery = json.loads(bridge.stdout.readline())
146+ request = {"version": 1, "id": 1, "op": "capture", "attachment": discovery["result"]["attachment"]}
147+ output, error = bridge.communicate(json.dumps(request) + "\n", timeout=1)
148+ envelopes = [json.loads(line) for line in output.splitlines()]
149+ if not envelopes or "error" in envelopes[-1]:
150+ raise RuntimeError("observation capture failed: " + str(envelopes) + error)
151+ return envelopes[-1]["result"]
152+ finally:
153+ if bridge.poll() is None:
154+ bridge.kill()
155+ bridge.communicate()
156+ compositor.stdin.write("sleep 1s\nkey Return\n")
157+ compositor.stdin.flush()
158+ deadline = time.monotonic() + 30
159+ while True:
160+ try:
161+ state = capture()
162+ if state.get("context", {}).get("status") == "attached" and state.get("context", {}).get("session", {}).get("session_name") == "resilience":
163+ break
164+ except (subprocess.SubprocessError, ValueError, RuntimeError) as error:
165+ (artifacts / "capture-error.txt").write_text(str(error) + "\n" + str(getattr(error, "stderr", "")))
166+ if time.monotonic() > deadline:
167+ diagnosis = subprocess.run(["docker", "exec", client, "sh", "-c", "ps -ef; ls -l /home/demo/visual/ui.sock"], capture_output=True, text=True, timeout=5)
168+ (artifacts / "client-processes.txt").write_text(diagnosis.stdout + diagnosis.stderr)
169+ (artifacts / "relay-stats.json").write_text(json.dumps(control("stats")))
170+ (artifacts / "last-capture.json").write_text(json.dumps(locals().get("state")))
171+ raise RuntimeError("visual client did not attach")
172+ time.sleep(0.1)
173+ (artifacts / "initial.json").write_text(json.dumps(state))
174+ def key(command):
175+ compositor.stdin.write(command + "\n")
176+ compositor.stdin.flush()
177+ def responsive(label):
178+ latest = shots / "latest.png"
179+ before = artifacts / (label + "-before.png")
180+ shutil.copyfile(latest, before)
181+ start = time.monotonic()
182+ stamp = latest.stat().st_mtime_ns
183+ key("key Alt+space")
184+ key("type SSP")
185+ key("key Return")
186+ while time.monotonic() - start < 2:
187+ if latest.stat().st_mtime_ns != stamp:
188+ after = artifacts / (label + "-after.png")
189+ shutil.copyfile(latest, after)
190+ changed = subprocess.run([str(ROOT / "build/resilience-pixels"), str(before), str(after)], capture_output=True, text=True)
191+ observed = capture()
192+ (artifacts / (label + "-observed.json")).write_text(json.dumps(observed))
193+ if changed.returncode == 0 and "Sessions" in observed.get("text", "") and "Esc close" in observed.get("text", ""):
194+ key("key Escape")
195+ return {"elapsed": time.monotonic()-start, "pixels": json.loads(changed.stdout), "context": observed["context"]}
196+ time.sleep(0.05)
197+ raise RuntimeError(label + ": local picker failed 2s response gate")
198+ evidence = {"healthy": responsive("healthy")}
199+ time.sleep(0.5)
200+ key("key Alt+space")
201+ key("type SSP")
202+ key("key Return")
203+ deadline = time.monotonic() + 5
204+ while "Sessions" not in capture().get("text", ""):
205+ if time.monotonic() > deadline:
206+ raise RuntimeError("picker did not open before blackout")
207+ time.sleep(0.1)
208+ before = artifacts / "blackout-before.png"
209+ shutil.copyfile(shots / "latest.png", before)
210+ assert "Sessions" in capture().get("text", ""), "picker was not open before blackout"
211+ impairment = os.environ.get("VEV_RESILIENCE_LINK", "both")
212+ if impairment not in ("both", "udp", "tcp"):
213+ raise ValueError("VEV_RESILIENCE_LINK must be both, udp or tcp")
214+ control("blackout", udp=impairment != "tcp", tcp=impairment != "udp", active=True)
215+ start = time.monotonic()
216+ key("type /resilience")
217+ deadline = start + 2
218+ while time.monotonic() < deadline:
219+ after = artifacts / "blackout-after.png"
220+ shutil.copyfile(shots / "latest.png", after)
221+ changed = subprocess.run([str(ROOT / "build/resilience-pixels"), str(before), str(after)], capture_output=True, text=True)
222+ observed = capture()
223+ if changed.returncode == 0 and "1 matches" in observed.get("text", ""):
224+ evidence["blackout"] = {"elapsed": time.monotonic()-start, "pixels": json.loads(changed.stdout)}
225+ break
226+ time.sleep(0.05)
227+ else:
228+ raise RuntimeError("open picker search froze during blackout")
229+ time.sleep(3)
230+ control("blackout", udp=True, tcp=True, active=False)
231+ key("key Escape")
232+ time.sleep(0.2)
233+ key("key Escape")
234+ deadline = time.monotonic() + 5
235+ while "Sessions" in capture().get("text", ""):
236+ if time.monotonic() > deadline:
237+ raise RuntimeError("picker did not close after restoration")
238+ time.sleep(0.1)
239+ key("key Ctrl+c")
240+ time.sleep(1)
241+ initial_session = state["context"]["session"]
242+ key("type printf 'RESILIENCE_RECOVERED\\n'")
243+ key("key Return")
244+ deadline = time.monotonic() + 10
245+ while time.monotonic() < deadline:
246+ recovered = capture()
247+ if "\nRESILIENCE_RECOVERED" in recovered.get("text", "") and recovered["context"]["session"] == initial_session:
248+ evidence["recovered"] = recovered["context"]
249+ break
250+ time.sleep(0.1)
251+ else:
252+ raise RuntimeError("session did not recover after blackout")
253+ # Negative control: compositor can still render while client is
254+ # stopped. No client observation is allowed to satisfy the gate.
255+ pid_result = docker("exec", client, "pgrep", "-f", "^vev attach", capture_output=True, text=True)
256+ pid = pid_result.stdout.strip().splitlines()[0]
257+ docker("exec", client, "kill", "-STOP", pid)
258+ frozen = artifacts / "frozen.png"
259+ shutil.copyfile(shots / "latest.png", frozen)
260+ try:
261+ key("key Alt+space")
262+ time.sleep(0.3)
263+ check = subprocess.run([str(ROOT / "build/resilience-pixels"), str(frozen), str(shots / "latest.png")], capture_output=True, text=True)
264+ assert check.returncode == 1, "visual gate accepted a stopped client"
265+ evidence["negative_control"] = "stopped client rejected"
266+ finally:
267+ docker("exec", client, "kill", "-CONT", pid)
268+ if os.environ.get("VEV_RESILIENCE_LONG") == "1":
269+ # The negative control's Alt+space was buffered while the
270+ # client was stopped and opens the palette once it resumes.
271+ # Close it on observed state only: a blind extra Escape would
272+ # reach the shell as a lone ESC, which readline keeps as a
273+ # Meta prefix and uses to eat the first replayed key.
274+ deadline = time.monotonic() + 5
275+ while "Commands" not in capture().get("text", ""):
276+ if time.monotonic() > deadline:
277+ raise RuntimeError("palette did not open after the stopped client resumed")
278+ time.sleep(0.1)
279+ key("key Escape")
280+ deadline = time.monotonic() + 5
281+ while "Commands" in capture().get("text", ""):
282+ if time.monotonic() > deadline:
283+ raise RuntimeError("palette did not close before the long outage")
284+ time.sleep(0.1)
285+ time.sleep(0.5)
286+ control("blackout", udp=True, tcp=True, active=True)
287+ if transport == "ssh":
288+ control("disconnect")
289+ deadline = time.monotonic() + 90
290+ while time.monotonic() < deadline:
291+ observed = capture()
292+ if observed.get("context", {}).get("status") == "connecting":
293+ break
294+ time.sleep(0.2)
295+ else:
296+ raise RuntimeError("long outage did not enter resume")
297+ if os.environ.get("VEV_RESILIENCE_RESUME") == "1":
298+ key("type echo HELD_ONE; echo HELD_TWO")
299+ key("key Return")
300+ # Keep the route down until the compositor has delivered
301+ # the complete physical input script to the resume owner.
302+ time.sleep(2)
303+ control("blackout", udp=True, tcp=True, active=False)
304+ deadline = time.monotonic() + 30
305+ while time.monotonic() < deadline:
306+ resumed = capture()
307+ text = resumed.get("text", "")
308+ if "\nHELD_ONE" in text and "\nHELD_TWO" in text and resumed.get("context", {}).get("session") == initial_session:
309+ assert text.index("\nHELD_ONE") < text.index("\nHELD_TWO"), "held input reordered"
310+ # The echoed command line proves the first held
311+ # byte was not swallowed before the shell saw it.
312+ assert "$ echo HELD_ONE; echo HELD_TWO" in text, "held input replayed incompletely"
313+ evidence["resumed"] = resumed["context"]
314+ break
315+ time.sleep(0.1)
316+ else:
317+ raise RuntimeError("long outage did not replay held input")
318+ else:
319+ start = time.monotonic()
320+ cancel_key = os.environ.get("VEV_RESILIENCE_CANCEL", "Ctrl+c")
321+ if cancel_key not in ("Ctrl+c", "Escape"):
322+ raise ValueError("cancel must be Ctrl+c or Escape")
323+ key("key " + cancel_key)
324+ deadline = start + 2
325+ while time.monotonic() < deadline:
326+ observed = capture()
327+ if observed.get("context", {}).get("status") == "picker":
328+ evidence["resume_cancel"] = {"elapsed": time.monotonic()-start}
329+ break
330+ time.sleep(0.05)
331+ else:
332+ raise RuntimeError(cancel_key + " did not cancel resume")
333+ control("blackout", udp=True, tcp=True, active=False)
334+ evidence["stats"] = control("stats")
335+ if transport == "quic":
336+ assert evidence["stats"]["udp"], "QUIC bypassed impairment relay"
337+ assert evidence["stats"]["tcp_bytes"] > 0, "SSH bypassed impairment relay"
338+ (artifacts / "evidence.json").write_text(json.dumps(evidence, indent=2))
339+ control("stop")
340+ finally:
341+ for process in reversed(processes):
342+ if process.poll() is None:
343+ process.terminate()
344+ try:
345+ process.wait(timeout=5)
346+ except subprocess.TimeoutExpired:
347+ process.kill()
348+ process.wait()
349+ for container in containers:
350+ role = container.rsplit("-", 1)[-1]
351+ for filename in ("vev-client.log", "vev-daemon.log", "broker/log/vev-broker.log"):
352+ subprocess.run(["docker", "cp", container + ":/home/demo/.local/state/vev/" + filename, str(artifacts / (role + "-" + filename.replace("/", "-")))], timeout=30, capture_output=True)
353+ subprocess.run(["docker", "rm", "-f", container], timeout=30, capture_output=True)
354+ if network:
355+ subprocess.run(["docker", "network", "rm", name], timeout=30, capture_output=True)
356+ subprocess.run(["docker", "image", "rm", name], timeout=30, capture_output=True)
357+ print("artifacts:", artifacts)
358+
359+
360+if __name__ == "__main__":
361+ main()
@@ -0,0 +1,361 @@
1+#!/usr/bin/env python3
2+"""Disposable dual-transport fixture and physical Wayland input acceptance."""
3+import importlib.util
4+import json
5+import os
6+from pathlib import Path
7+import shutil
8+import signal
9+import subprocess
10+import tempfile
11+import time
12+import uuid
13+
14+ROOT = Path(__file__).resolve().parents[2]
15+
16+
17+def run(*args, **kwargs):
18+ return subprocess.run(args, check=True, timeout=kwargs.pop("timeout", 60), **kwargs)
19+
20+
21+def docker(*args, **kwargs):
22+ return run("docker", *args, **kwargs)
23+
24+
25+def main():
26+ name = "vev-resilience-" + uuid.uuid4().hex[:10]
27+ artifacts = Path(tempfile.mkdtemp(prefix="vev-resilience-artifacts-"))
28+ processes = []
29+ containers = []
30+ network = False
31+ def interrupt(signum, frame):
32+ raise KeyboardInterrupt("fixture interrupted")
33+ signal.signal(signal.SIGTERM, interrupt)
34+ with tempfile.TemporaryDirectory(prefix="vev-resilience-private-") as directory:
35+ private = Path(directory)
36+ runtime = private / "runtime"
37+ runtime.mkdir(mode=0o700)
38+ try:
39+ for tool in ("docker", "foot", "ssh-keygen", "go"):
40+ if shutil.which(tool) is None:
41+ raise RuntimeError("missing prerequisite: " + tool)
42+ run("go", "build", "-o", str(ROOT / "build/vev"), ".", cwd=ROOT)
43+ for helper in ("relay", "pixels"):
44+ run("go", "build", "-o", str(ROOT / f"build/resilience-{helper}"), f"./scripts/remote-resilience-harness/{helper}", cwd=ROOT)
45+ nefer = os.environ.get("NEFERWL_BIN")
46+ if not nefer:
47+ nefer = str(private / "neferwl")
48+ common = run("git", "rev-parse", "--path-format=absolute", "--git-common-dir", cwd=ROOT, capture_output=True, text=True).stdout.strip()
49+ source = Path(os.environ.get("NEFERWL_SOURCE", str(Path(common).parent.parent / "neferwl")))
50+ run("go", "build", "-o", nefer, "./cmd/neferwl", cwd=source, env={**os.environ, "CGO_ENABLED": "0"})
51+ docker("build", "-f", str(ROOT / "scripts/demo/Dockerfile"), "-t", name, str(ROOT), timeout=600)
52+ docker("network", "create", name)
53+ network = True
54+ for role in ("client", "remote"):
55+ container = name + "-" + role
56+ docker("run", "-d", "--name", container, "--network", name, "--network-alias", "upstream" if role == "remote" else "remote", "--entrypoint", "sleep", name, "infinity")
57+ containers.append(container)
58+ client, remote = containers
59+ for key in ("client", "host"):
60+ run("ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "fixture", "-f", str(private / key))
61+ docker("cp", str(private / "host"), remote + ":/etc/ssh/ssh_host_ed25519_key")
62+ docker("cp", str(private / "host.pub"), remote + ":/etc/ssh/ssh_host_ed25519_key.pub")
63+ docker("cp", str(private / "client.pub"), remote + ":/home/demo/.ssh/authorized_keys")
64+ docker("exec", "--user", "root", remote, "sh", "-c", "rm -f /home/demo/.ssh/id_ed25519*; chmod 600 /etc/ssh/ssh_host_ed25519_key /home/demo/.ssh/authorized_keys; chown demo:demo /home/demo/.ssh/authorized_keys; mkdir -p /run/sshd; /usr/sbin/sshd")
65+ docker("exec", "--user", "root", client, "sh", "-c", "rm -f /home/demo/.ssh/id_ed25519*")
66+ docker("cp", str(private / "client"), client + ":/home/demo/.ssh/id_ed25519")
67+ public = (private / "host.pub").read_text().split()
68+ (private / "known_hosts").write_text("remote " + " ".join(public[:2]) + "\n")
69+ (private / "config").write_text("Host remote\n HostName remote\n Port 2222\n User demo\n IdentityFile ~/.ssh/id_ed25519\n IdentitiesOnly yes\n StrictHostKeyChecking yes\n HostKeyAlias remote\n UserKnownHostsFile ~/.ssh/known_hosts\n")
70+ for file in ("known_hosts", "config"):
71+ docker("cp", str(private / file), client + ":/home/demo/.ssh/" + file)
72+ docker("cp", str(ROOT / "build/resilience-relay"), client + ":/tmp/relay")
73+ docker("cp", str(ROOT / "scripts/remote-resilience-harness/ssh.py"), client + ":/tmp/ssh")
74+ docker("exec", "--user", "root", client, "sh", "-c", "chown demo:demo /home/demo/.ssh/*; chmod 600 /home/demo/.ssh/*; chmod 755 /tmp/ssh; install -d -m 700 -o demo -g demo /home/demo/relay; mkdir -p /tmp/bin; ln -s /tmp/ssh /tmp/bin/ssh; mv /usr/bin/ssh /usr/bin/ssh-real; ln -s /tmp/ssh /usr/bin/ssh")
75+ transport = os.environ.get("VEV_RESILIENCE_TRANSPORT", "quic")
76+ if transport not in ("quic", "ssh"):
77+ raise ValueError("VEV_RESILIENCE_TRANSPORT must be quic or ssh")
78+ relay_args = ["-degraded"] if os.environ.get("VEV_RESILIENCE_DEGRADED") == "1" else []
79+ relay = subprocess.Popen(["docker", "exec", client, "/tmp/relay", "-remote", "upstream", "-control", "/home/demo/relay/control.sock", *relay_args])
80+ processes.append(relay)
81+ control_id = 0
82+ def control(op, **fields):
83+ nonlocal control_id
84+ control_id += 1
85+ request = {"id": control_id, "op": op, **fields}
86+ code = "import socket,json; s=socket.socket(socket.AF_UNIX); s.settimeout(3); s.connect('/home/demo/relay/control.sock'); s.sendall(" + repr(json.dumps(request).encode() + b"\n") + "); print(s.makefile().readline())"
87+ result = docker("exec", client, "python3", "-c", code, capture_output=True, text=True)
88+ reply = json.loads(result.stdout)
89+ assert reply["id"] == control_id and "error" not in reply, reply
90+ return reply["result"]
91+ # Fixture image must explicitly supply Python rather than using host tools.
92+ docker("exec", "--user", "root", client, "sh", "-c", "command -v python3", capture_output=True)
93+ deadline = time.monotonic() + 5
94+ while True:
95+ ready = subprocess.run(["docker", "exec", client, "test", "-S", "/home/demo/relay/control.sock"], capture_output=True, timeout=3)
96+ if ready.returncode == 0:
97+ break
98+ if time.monotonic() > deadline:
99+ raise RuntimeError("relay control socket did not become ready")
100+ time.sleep(0.05)
101+ probe = subprocess.run(["docker", "exec", client, "/usr/bin/ssh", "-v", "remote", "true"], capture_output=True, text=True, timeout=10)
102+ if probe.returncode:
103+ (artifacts / "ssh-probe.log").write_text(probe.stderr)
104+ raise RuntimeError("fixture SSH authentication failed")
105+ env_args = ["-e", "PATH=/tmp/bin:/usr/local/bin:/usr/bin", "-e", "VEV_LOG=debug", "-e", "SSH_AUTH_SOCK="]
106+ spec = importlib.util.spec_from_file_location("acceptance", ROOT / "scripts/remote-picker-harness/acceptance.py")
107+ acceptance = importlib.util.module_from_spec(spec)
108+ spec.loader.exec_module(acceptance)
109+ fixture = acceptance.Driver(remote, ["--session", "resilience"])
110+ fixture.close()
111+ docker("exec", *env_args, client, "vev", "host", "add", "--transport", transport, "remote")
112+ # Wait for the broker publication before resolving direct attach.
113+ deadline = time.monotonic() + 30
114+ while True:
115+ inventory = docker("exec", *env_args, client, "vev", "ls", "remote", capture_output=True, text=True)
116+ if "resilience" in inventory.stdout:
117+ break
118+ (artifacts / "inventory.txt").write_text(inventory.stdout + inventory.stderr)
119+ if time.monotonic() > deadline:
120+ raise RuntimeError("remote fixture was not published")
121+ time.sleep(0.2)
122+ shots = artifacts / "shots"
123+ shots.mkdir()
124+ docker_endpoint = run("docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}", capture_output=True, text=True).stdout.strip()
125+ env = {**os.environ, "DOCKER_HOST": docker_endpoint, "XDG_RUNTIME_DIR": str(runtime), "XDG_CONFIG_HOME": str(private / "config-home"), "XDG_STATE_HOME": str(private / "state"), "HOME": str(private)}
126+ env["TERMINAL"] = "foot --config=/dev/null --override=cursor.blink=no --log-level=info --hold -- docker --host=" + docker_endpoint + " exec -it " + " ".join(env_args) + " " + client + " vev attach demo@remote:resilience --ui-observe --ui-socket /home/demo/visual/ui.sock"
127+ log = (artifacts / "compositor.log").open("w")
128+ compositor = subprocess.Popen([nefer, "--backend=headless", "--size=1280x720", "--input=-", "--screenshot", str(shots), "--timeout", "180s"], stdin=subprocess.PIPE, stdout=log, stderr=log, text=True, env=env)
129+ processes.append(compositor)
130+ deadline = time.monotonic() + 15
131+ while not list(runtime.glob("wayland-*")):
132+ if compositor.poll() is not None or time.monotonic() > deadline:
133+ raise RuntimeError("headless compositor did not create socket")
134+ time.sleep(0.05)
135+ sockets = [p for p in runtime.glob("wayland-*") if not p.name.endswith(".lock")]
136+ env["WAYLAND_DISPLAY"] = str(sockets[0])
137+ env.pop("WAYLAND_SOCKET", None)
138+ env.pop("DISPLAY", None)
139+ def capture():
140+ bridge = subprocess.Popen(["docker", "exec", "-i", client, "vev", "--ui-driver", "--socket", "/home/demo/visual/ui.sock"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
141+ try:
142+ import select
143+ if not select.select([bridge.stdout], [], [], 1)[0]:
144+ raise RuntimeError("observation discovery timed out")
145+ discovery = json.loads(bridge.stdout.readline())
146+ request = {"version": 1, "id": 1, "op": "capture", "attachment": discovery["result"]["attachment"]}
147+ output, error = bridge.communicate(json.dumps(request) + "\n", timeout=1)
148+ envelopes = [json.loads(line) for line in output.splitlines()]
149+ if not envelopes or "error" in envelopes[-1]:
150+ raise RuntimeError("observation capture failed: " + str(envelopes) + error)
151+ return envelopes[-1]["result"]
152+ finally:
153+ if bridge.poll() is None:
154+ bridge.kill()
155+ bridge.communicate()
156+ compositor.stdin.write("sleep 1s\nkey Return\n")
157+ compositor.stdin.flush()
158+ deadline = time.monotonic() + 30
159+ while True:
160+ try:
161+ state = capture()
162+ if state.get("context", {}).get("status") == "attached" and state.get("context", {}).get("session", {}).get("session_name") == "resilience":
163+ break
164+ except (subprocess.SubprocessError, ValueError, RuntimeError) as error:
165+ (artifacts / "capture-error.txt").write_text(str(error) + "\n" + str(getattr(error, "stderr", "")))
166+ if time.monotonic() > deadline:
167+ diagnosis = subprocess.run(["docker", "exec", client, "sh", "-c", "ps -ef; ls -l /home/demo/visual/ui.sock"], capture_output=True, text=True, timeout=5)
168+ (artifacts / "client-processes.txt").write_text(diagnosis.stdout + diagnosis.stderr)
169+ (artifacts / "relay-stats.json").write_text(json.dumps(control("stats")))
170+ (artifacts / "last-capture.json").write_text(json.dumps(locals().get("state")))
171+ raise RuntimeError("visual client did not attach")
172+ time.sleep(0.1)
173+ (artifacts / "initial.json").write_text(json.dumps(state))
174+ def key(command):
175+ compositor.stdin.write(command + "\n")
176+ compositor.stdin.flush()
177+ def responsive(label):
178+ latest = shots / "latest.png"
179+ before = artifacts / (label + "-before.png")
180+ shutil.copyfile(latest, before)
181+ start = time.monotonic()
182+ stamp = latest.stat().st_mtime_ns
183+ key("key Alt+space")
184+ key("type SSP")
185+ key("key Return")
186+ while time.monotonic() - start < 2:
187+ if latest.stat().st_mtime_ns != stamp:
188+ after = artifacts / (label + "-after.png")
189+ shutil.copyfile(latest, after)
190+ changed = subprocess.run([str(ROOT / "build/resilience-pixels"), str(before), str(after)], capture_output=True, text=True)
191+ observed = capture()
192+ (artifacts / (label + "-observed.json")).write_text(json.dumps(observed))
193+ if changed.returncode == 0 and "Sessions" in observed.get("text", "") and "Esc close" in observed.get("text", ""):
194+ key("key Escape")
195+ return {"elapsed": time.monotonic()-start, "pixels": json.loads(changed.stdout), "context": observed["context"]}
196+ time.sleep(0.05)
197+ raise RuntimeError(label + ": local picker failed 2s response gate")
198+ evidence = {"healthy": responsive("healthy")}
199+ time.sleep(0.5)
200+ key("key Alt+space")
201+ key("type SSP")
202+ key("key Return")
203+ deadline = time.monotonic() + 5
204+ while "Sessions" not in capture().get("text", ""):
205+ if time.monotonic() > deadline:
206+ raise RuntimeError("picker did not open before blackout")
207+ time.sleep(0.1)
208+ before = artifacts / "blackout-before.png"
209+ shutil.copyfile(shots / "latest.png", before)
210+ assert "Sessions" in capture().get("text", ""), "picker was not open before blackout"
211+ impairment = os.environ.get("VEV_RESILIENCE_LINK", "both")
212+ if impairment not in ("both", "udp", "tcp"):
213+ raise ValueError("VEV_RESILIENCE_LINK must be both, udp or tcp")
214+ control("blackout", udp=impairment != "tcp", tcp=impairment != "udp", active=True)
215+ start = time.monotonic()
216+ key("type /resilience")
217+ deadline = start + 2
218+ while time.monotonic() < deadline:
219+ after = artifacts / "blackout-after.png"
220+ shutil.copyfile(shots / "latest.png", after)
221+ changed = subprocess.run([str(ROOT / "build/resilience-pixels"), str(before), str(after)], capture_output=True, text=True)
222+ observed = capture()
223+ if changed.returncode == 0 and "1 matches" in observed.get("text", ""):
224+ evidence["blackout"] = {"elapsed": time.monotonic()-start, "pixels": json.loads(changed.stdout)}
225+ break
226+ time.sleep(0.05)
227+ else:
228+ raise RuntimeError("open picker search froze during blackout")
229+ time.sleep(3)
230+ control("blackout", udp=True, tcp=True, active=False)
231+ key("key Escape")
232+ time.sleep(0.2)
233+ key("key Escape")
234+ deadline = time.monotonic() + 5
235+ while "Sessions" in capture().get("text", ""):
236+ if time.monotonic() > deadline:
237+ raise RuntimeError("picker did not close after restoration")
238+ time.sleep(0.1)
239+ key("key Ctrl+c")
240+ time.sleep(1)
241+ initial_session = state["context"]["session"]
242+ key("type printf 'RESILIENCE_RECOVERED\\n'")
243+ key("key Return")
244+ deadline = time.monotonic() + 10
245+ while time.monotonic() < deadline:
246+ recovered = capture()
247+ if "\nRESILIENCE_RECOVERED" in recovered.get("text", "") and recovered["context"]["session"] == initial_session:
248+ evidence["recovered"] = recovered["context"]
249+ break
250+ time.sleep(0.1)
251+ else:
252+ raise RuntimeError("session did not recover after blackout")
253+ # Negative control: compositor can still render while client is
254+ # stopped. No client observation is allowed to satisfy the gate.
255+ pid_result = docker("exec", client, "pgrep", "-f", "^vev attach", capture_output=True, text=True)
256+ pid = pid_result.stdout.strip().splitlines()[0]
257+ docker("exec", client, "kill", "-STOP", pid)
258+ frozen = artifacts / "frozen.png"
259+ shutil.copyfile(shots / "latest.png", frozen)
260+ try:
261+ key("key Alt+space")
262+ time.sleep(0.3)
263+ check = subprocess.run([str(ROOT / "build/resilience-pixels"), str(frozen), str(shots / "latest.png")], capture_output=True, text=True)
264+ assert check.returncode == 1, "visual gate accepted a stopped client"
265+ evidence["negative_control"] = "stopped client rejected"
266+ finally:
267+ docker("exec", client, "kill", "-CONT", pid)
268+ if os.environ.get("VEV_RESILIENCE_LONG") == "1":
269+ # The negative control's Alt+space was buffered while the
270+ # client was stopped and opens the palette once it resumes.
271+ # Close it on observed state only: a blind extra Escape would
272+ # reach the shell as a lone ESC, which readline keeps as a
273+ # Meta prefix and uses to eat the first replayed key.
274+ deadline = time.monotonic() + 5
275+ while "Commands" not in capture().get("text", ""):
276+ if time.monotonic() > deadline:
277+ raise RuntimeError("palette did not open after the stopped client resumed")
278+ time.sleep(0.1)
279+ key("key Escape")
280+ deadline = time.monotonic() + 5
281+ while "Commands" in capture().get("text", ""):
282+ if time.monotonic() > deadline:
283+ raise RuntimeError("palette did not close before the long outage")
284+ time.sleep(0.1)
285+ time.sleep(0.5)
286+ control("blackout", udp=True, tcp=True, active=True)
287+ if transport == "ssh":
288+ control("disconnect")
289+ deadline = time.monotonic() + 90
290+ while time.monotonic() < deadline:
291+ observed = capture()
292+ if observed.get("context", {}).get("status") == "connecting":
293+ break
294+ time.sleep(0.2)
295+ else:
296+ raise RuntimeError("long outage did not enter resume")
297+ if os.environ.get("VEV_RESILIENCE_RESUME") == "1":
298+ key("type echo HELD_ONE; echo HELD_TWO")
299+ key("key Return")
300+ # Keep the route down until the compositor has delivered
301+ # the complete physical input script to the resume owner.
302+ time.sleep(2)
303+ control("blackout", udp=True, tcp=True, active=False)
304+ deadline = time.monotonic() + 30
305+ while time.monotonic() < deadline:
306+ resumed = capture()
307+ text = resumed.get("text", "")
308+ if "\nHELD_ONE" in text and "\nHELD_TWO" in text and resumed.get("context", {}).get("session") == initial_session:
309+ assert text.index("\nHELD_ONE") < text.index("\nHELD_TWO"), "held input reordered"
310+ # The echoed command line proves the first held
311+ # byte was not swallowed before the shell saw it.
312+ assert "$ echo HELD_ONE; echo HELD_TWO" in text, "held input replayed incompletely"
313+ evidence["resumed"] = resumed["context"]
314+ break
315+ time.sleep(0.1)
316+ else:
317+ raise RuntimeError("long outage did not replay held input")
318+ else:
319+ start = time.monotonic()
320+ cancel_key = os.environ.get("VEV_RESILIENCE_CANCEL", "Ctrl+c")
321+ if cancel_key not in ("Ctrl+c", "Escape"):
322+ raise ValueError("cancel must be Ctrl+c or Escape")
323+ key("key " + cancel_key)
324+ deadline = start + 2
325+ while time.monotonic() < deadline:
326+ observed = capture()
327+ if observed.get("context", {}).get("status") == "picker":
328+ evidence["resume_cancel"] = {"elapsed": time.monotonic()-start}
329+ break
330+ time.sleep(0.05)
331+ else:
332+ raise RuntimeError(cancel_key + " did not cancel resume")
333+ control("blackout", udp=True, tcp=True, active=False)
334+ evidence["stats"] = control("stats")
335+ if transport == "quic":
336+ assert evidence["stats"]["udp"], "QUIC bypassed impairment relay"
337+ assert evidence["stats"]["tcp_bytes"] > 0, "SSH bypassed impairment relay"
338+ (artifacts / "evidence.json").write_text(json.dumps(evidence, indent=2))
339+ control("stop")
340+ finally:
341+ for process in reversed(processes):
342+ if process.poll() is None:
343+ process.terminate()
344+ try:
345+ process.wait(timeout=5)
346+ except subprocess.TimeoutExpired:
347+ process.kill()
348+ process.wait()
349+ for container in containers:
350+ role = container.rsplit("-", 1)[-1]
351+ for filename in ("vev-client.log", "vev-daemon.log", "broker/log/vev-broker.log"):
352+ subprocess.run(["docker", "cp", container + ":/home/demo/.local/state/vev/" + filename, str(artifacts / (role + "-" + filename.replace("/", "-")))], timeout=30, capture_output=True)
353+ subprocess.run(["docker", "rm", "-f", container], timeout=30, capture_output=True)
354+ if network:
355+ subprocess.run(["docker", "network", "rm", name], timeout=30, capture_output=True)
356+ subprocess.run(["docker", "image", "rm", name], timeout=30, capture_output=True)
357+ print("artifacts:", artifacts)
358+
359+
360+if __name__ == "__main__":
361+ main()

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close