fix: resolve reply-bearing X11 liveness cookies with Reply (#2)

cmd/groundstation-display/main.go

@@ -44,6 +44,28 @@ func buildTarget(name string) ([]display.Layer, error) {
 	}
 }
 
+// wireDiagnostics gives every layer the shared structured reconnect hook so
+// each layer reports its own reconnect boundaries, identified by target and
+// layer name.
+func wireDiagnostics(target string, layers []display.Layer) {
+	for i := range layers {
+		layers[i].Options = display.Options{
+			Target: target,
+			Layer:  layers[i].Name,
+			Diagnostic: func(d display.Diagnostic) {
+				log.Printf("display reconnect: %s", formatDiagnostic(d))
+			},
+		}
+	}
+}
+
+// formatDiagnostic renders one reconnect diagnostic as key=value fields so
+// reconnects stay greppable and machine-parseable in the journal.
+func formatDiagnostic(d display.Diagnostic) string {
+	return fmt.Sprintf("target=%s layer=%s event=%s error=%q retry_delay=%s",
+		d.Target, d.Layer, d.Event, d.Err, d.RetryDelay)
+}
+
 func main() {
 	target := flag.String("target", "x11", "display target: x11 (Viture) today; mipi planned")
 	flag.Parse()
@@ -52,11 +74,13 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
+	wireDiagnostics(*target, layers)
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 
 	// Temporary X11 unavailability never exits this process; every layer
-	// keeps retrying internally until the context is cancelled.
+	// keeps retrying internally until the context is cancelled, reporting
+	// each reconnect boundary through the structured diagnostic hook.
 	display.RunLayers(ctx, layers)
 }

cmd/groundstation-display/main_test.go

@@ -1,7 +1,9 @@
 package main
 
 import (
+	"errors"
 	"testing"
+	"time"
 
 	"git.piksar.eu/piotr/groundstation/internal/display"
 )
@@ -83,3 +85,41 @@ func TestBuildTargetRejectsUnknownTarget(t *testing.T) {
 		t.Fatal("unknown target must fail")
 	}
 }
+
+func TestFormatDiagnosticRendersStructuredFields(t *testing.T) {
+	got := formatDiagnostic(display.Diagnostic{
+		Target:     "x11",
+		Layer:      "fallback-canvas",
+		Event:      display.EventOpenFailed,
+		Err:        errors.New("connect X11: refused"),
+		RetryDelay: 500 * time.Millisecond,
+	})
+	want := `target=x11 layer=fallback-canvas event=open_failed error="connect X11: refused" retry_delay=500ms`
+	if got != want {
+		t.Fatalf("formatDiagnostic=%q want %q", got, want)
+	}
+}
+
+func TestWireDiagnosticsTagsEveryLayerWithItsName(t *testing.T) {
+	layers, err := buildTarget("x11")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	wireDiagnostics("x11", layers)
+
+	for _, layer := range layers {
+		if layer.Options.Target != "x11" {
+			t.Fatalf("layer %q target=%q want x11", layer.Name, layer.Options.Target)
+		}
+		if layer.Options.Layer != layer.Name {
+			t.Fatalf("layer %q diagnostic layer=%q", layer.Name, layer.Options.Layer)
+		}
+		if layer.Options.Diagnostic == nil {
+			t.Fatalf("layer %q must carry the reconnect diagnostic hook", layer.Name)
+		}
+	}
+
+	// The shared hook must be safe to call from any layer goroutine.
+	layers[0].Options.Diagnostic(display.Diagnostic{Target: "x11", Layer: layers[0].Name})
+}

internal/display/diagnostics_test.go

@@ -0,0 +1,182 @@
+package display
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+// diagnosticCapture collects structured reconnect records from Run.
+type diagnosticCapture struct {
+	mu      sync.Mutex
+	records []Diagnostic
+}
+
+func (c *diagnosticCapture) record(d Diagnostic) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.records = append(c.records, d)
+}
+
+func (c *diagnosticCapture) all() []Diagnostic {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return append([]Diagnostic(nil), c.records...)
+}
+
+func TestRunDiagnosesEachFailedOpenWithStructuredFields(t *testing.T) {
+	session := newFakeSession()
+	var opens int
+	opener := fakeOpener{open: func(int) (Session, error) {
+		opens++
+		if opens <= 2 {
+			return nil, errors.New("x11 down")
+		}
+		return session, nil
+	}}
+
+	capture := &diagnosticCapture{}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() {
+		done <- Run(ctx, opener, NewStatusTemplate("hello"), Options{
+			Target:     "x11",
+			Layer:      "fallback-canvas",
+			Diagnostic: capture.record,
+			RetryDelay: time.Millisecond,
+			MaxDelay:   2 * time.Millisecond,
+		})
+	}()
+
+	waitFor(t, func() bool { return session.canvas.lineCount() > 0 }, "connect after retries")
+	cancel()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("run: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("run did not stop after cancel")
+	}
+
+	records := capture.all()
+	if len(records) != 2 {
+		t.Fatalf("records=%d want exactly one per failed open (2)", len(records))
+	}
+	first, second := records[0], records[1]
+	if first.Target != "x11" {
+		t.Fatalf("target=%q want x11", first.Target)
+	}
+	if first.Layer != "fallback-canvas" {
+		t.Fatalf("layer=%q want fallback-canvas", first.Layer)
+	}
+	if first.Event != EventOpenFailed {
+		t.Fatalf("event=%q want %q", first.Event, EventOpenFailed)
+	}
+	if first.Err == nil || first.Err.Error() != "x11 down" {
+		t.Fatalf("underlying error %v must be carried in the record", first.Err)
+	}
+	if first.RetryDelay != time.Millisecond {
+		t.Fatalf("first retry delay=%v want %v", first.RetryDelay, time.Millisecond)
+	}
+	if second.RetryDelay != 2*time.Millisecond {
+		t.Fatalf("backed-off retry delay=%v want %v", second.RetryDelay, 2*time.Millisecond)
+	}
+}
+
+func TestRunDiagnosesBrokenSessionOnceAndRecovers(t *testing.T) {
+	first := newFakeSession()
+	second := newFakeSession()
+	var opens int
+	opener := fakeOpener{open: func(int) (Session, error) {
+		opens++
+		if opens == 1 {
+			return first, nil
+		}
+		return second, nil
+	}}
+
+	capture := &diagnosticCapture{}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() {
+		done <- Run(ctx, opener, NewStatusTemplate("hello"), Options{
+			Target:     "x11",
+			Diagnostic: capture.record,
+			RetryDelay: time.Millisecond,
+			MaxDelay:   2 * time.Millisecond,
+		})
+	}()
+
+	waitFor(t, func() bool { return first.canvas.lineCount() == 1 }, "draw on first session")
+	first.events <- eventResult{err: errors.New("connection lost")}
+	waitFor(t, func() bool { return second.canvas.lineCount() == 1 }, "recover on reconnected session")
+	cancel()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("run: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("run did not stop after cancel")
+	}
+
+	records := capture.all()
+	if len(records) != 1 {
+		t.Fatalf("records=%d want exactly one per broken session (1)", len(records))
+	}
+	record := records[0]
+	if record.Target != "x11" {
+		t.Fatalf("target=%q want x11", record.Target)
+	}
+	if record.Event != EventSessionEnded {
+		t.Fatalf("event=%q want %q", record.Event, EventSessionEnded)
+	}
+	if record.Err == nil || record.Err.Error() != "connection lost" {
+		t.Fatalf("underlying error %v must be carried in the record", record.Err)
+	}
+	if record.RetryDelay != time.Millisecond {
+		t.Fatalf("retry delay=%v want the reset %v", record.RetryDelay, time.Millisecond)
+	}
+}
+
+// blockingOpener fails only after the context is done, so cancellation is
+// the sole reason Run stops: no diagnostic may fire for shutdown.
+type blockingOpener struct{}
+
+func (blockingOpener) Open(ctx context.Context) (Session, error) {
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func TestRunStaysSilentWhenCancellationEndsTheLoop(t *testing.T) {
+	capture := &diagnosticCapture{}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() {
+		done <- Run(ctx, blockingOpener{}, NewStatusTemplate("hello"), Options{
+			Target:     "x11",
+			Diagnostic: capture.record,
+			RetryDelay: time.Millisecond,
+			MaxDelay:   2 * time.Millisecond,
+		})
+	}()
+
+	cancel()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("run: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("run did not stop after cancel")
+	}
+	if records := capture.all(); len(records) != 0 {
+		t.Fatalf("records=%v must stay empty: shutdown is not a reconnect boundary", records)
+	}
+}

internal/display/display.go

@@ -49,12 +49,47 @@ type Opener interface {
 	Open(ctx context.Context) (Session, error)
 }
 
+// Reconnect-boundary diagnostic events.
+const (
+	// EventOpenFailed marks a failed attempt to open the display target.
+	EventOpenFailed = "open_failed"
+	// EventSessionEnded marks a session broken by a target failure.
+	EventSessionEnded = "session_ended"
+)
+
+// Diagnostic is one structured record from a reconnect boundary of the
+// kiosk loop. Exactly one record is emitted per failed open or broken
+// session; cancellation and shutdown stay silent.
+type Diagnostic struct {
+	// Target is the display target name, for example "x11".
+	Target string
+	// Layer is the display layer name within the target, for example
+	// "fallback-canvas"; empty for single-layer runs.
+	Layer string
+	// Event names the boundary: EventOpenFailed or EventSessionEnded.
+	Event string
+	// Err is the underlying error from Opener.Open or Session.WaitEvent.
+	Err error
+	// RetryDelay is the wait scheduled after this boundary.
+	RetryDelay time.Duration
+}
+
 // Options tune the kiosk lifecycle loop. Zero values select the defaults.
 type Options struct {
 	// RetryDelay is the first wait after a failed connect or lost session.
 	RetryDelay time.Duration
 	// MaxDelay bounds the exponential reconnect backoff.
 	MaxDelay time.Duration
+	// Target names the display target in diagnostics. Empty selects
+	// "unknown".
+	Target string
+	// Layer names the kiosk layer in diagnostics; empty for single-layer
+	// runs.
+	Layer string
+	// Diagnostic receives one structured record per reconnect boundary.
+	// Nil disables diagnostics. It runs on the loop goroutine and must not
+	// block.
+	Diagnostic func(Diagnostic)
 }
 
 func (o Options) withDefaults() Options {
@@ -67,10 +102,35 @@ func (o Options) withDefaults() Options {
 	return o
 }
 
+func (o Options) targetName() string {
+	if o.Target == "" {
+		return "unknown"
+	}
+	return o.Target
+}
+
+// diagnose emits one reconnect-boundary record. Cancellation is shutdown,
+// not a failure, so it stays silent; so does a nil hook.
+func (o Options) diagnose(ctx context.Context, event string, err error, retryDelay time.Duration) {
+	if o.Diagnostic == nil || err == nil || ctx.Err() != nil {
+		return
+	}
+	o.Diagnostic(Diagnostic{
+		Target:     o.targetName(),
+		Layer:      o.Layer,
+		Event:      event,
+		Err:        err,
+		RetryDelay: retryDelay,
+	})
+}
+
 // Run owns the kiosk display lifecycle until ctx is cancelled. Temporary
 // unavailability of the target, for example X11 not being up yet, never
 // returns an error: Run retries with bounded backoff and only stops when the
-// context is cancelled, so the Ground Station supervisor stays up.
+// context is cancelled, so the Ground Station supervisor stays up. Each
+// reconnect boundary (failed open, broken session) emits one structured
+// Diagnostic record through Options.Diagnostic, never through an error
+// return; cancellation stays silent.
 func Run(ctx context.Context, opener Opener, template Template, options Options) error {
 	opts := options.withDefaults()
 	delay := opts.RetryDelay
@@ -81,6 +141,7 @@ func Run(ctx context.Context, opener Opener, template Template, options Options)
 
 		session, err := opener.Open(ctx)
 		if err != nil {
+			opts.diagnose(ctx, EventOpenFailed, err, delay)
 			if !sleep(ctx, delay) {
 				return nil
 			}
@@ -91,9 +152,11 @@ func Run(ctx context.Context, opener Opener, template Template, options Options)
 		delay = opts.RetryDelay
 		canvas := session.Canvas()
 		template.Draw(canvas)
+		var sessionErr error
 		for ctx.Err() == nil {
-			redraw, err := session.WaitEvent(ctx)
-			if err != nil {
+			var redraw bool
+			redraw, sessionErr = session.WaitEvent(ctx)
+			if sessionErr != nil {
 				break
 			}
 			if redraw {
@@ -105,6 +168,7 @@ func Run(ctx context.Context, opener Opener, template Template, options Options)
 		if ctx.Err() != nil {
 			return nil
 		}
+		opts.diagnose(ctx, EventSessionEnded, sessionErr, delay)
 		if !sleep(ctx, delay) {
 			return nil
 		}

internal/display/liveness_test.go

@@ -0,0 +1,145 @@
+package display
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/BurntSushi/xgb"
+	"github.com/BurntSushi/xgb/xproto"
+)
+
+func TestConfirmMappedSucceedsWhenLivenessReplySucceeds(t *testing.T) {
+	// Open's post-map liveness stage must succeed when the reply-bearing
+	// round trip succeeds, so the mapped window proceeds to a live session.
+	probes := 0
+	probe := func() error {
+		probes++
+		return nil
+	}
+
+	if err := confirmMapped(probe); err != nil {
+		t.Fatalf("confirmMapped=%v want success when the liveness reply arrives", err)
+	}
+	if probes != 1 {
+		t.Fatalf("probes=%d want exactly one liveness round trip", probes)
+	}
+}
+
+func TestConfirmMappedWrapsLivenessFailureForMappedWindowCleanup(t *testing.T) {
+	sentinel := errors.New("connection reset")
+	err := confirmMapped(func() error { return sentinel })
+	if err == nil {
+		t.Fatal("confirmMapped must propagate a failed liveness round trip")
+	}
+	if !strings.Contains(err.Error(), "map kiosk window") {
+		t.Fatalf("error %v must be wrapped as the map-kiosk-window failure so Open closes the connection", err)
+	}
+	if !errors.Is(err, sentinel) {
+		t.Fatalf("error %v must carry the underlying liveness error", err)
+	}
+}
+
+func TestX11SessionWaitEventKeepsPollingWhileLivenessRepliesSucceed(t *testing.T) {
+	probes := 0
+	session := &x11Session{
+		poll: time.Millisecond,
+		pollEvent: func() (xgb.Event, xgb.Error) {
+			return nil, nil
+		},
+		probe: func() error {
+			probes++
+			return nil
+		},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+	_, err := session.WaitEvent(ctx)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("WaitEvent=%v want clean cancellation while liveness replies succeed", err)
+	}
+	if probes == 0 {
+		t.Fatal("WaitEvent must perform liveness round trips while idle")
+	}
+}
+
+func TestX11SessionWaitEventPropagatesLivenessError(t *testing.T) {
+	sentinel := errors.New("X server went away")
+	session := &x11Session{
+		poll:      time.Millisecond,
+		pollEvent: func() (xgb.Event, xgb.Error) { return nil, nil },
+		probe:     func() error { return sentinel },
+	}
+
+	// The propagated error is what makes Run close the mapped window and
+	// reconnect, keeping the failure contained in the display loop.
+	_, err := session.WaitEvent(context.Background())
+	if !errors.Is(err, sentinel) {
+		t.Fatalf("WaitEvent=%v want the underlying liveness error", err)
+	}
+}
+
+func TestX11SessionWaitEventStillHandlesEventsBeforeLiveness(t *testing.T) {
+	session := &x11Session{
+		poll: time.Millisecond,
+		pollEvent: func() (xgb.Event, xgb.Error) {
+			return xproto.ExposeEvent{}, nil
+		},
+		probe: func() error {
+			t.Fatal("liveness must not run when an event is already handled")
+			return nil
+		},
+	}
+
+	redraw, err := session.WaitEvent(context.Background())
+	if err != nil {
+		t.Fatalf("WaitEvent=%v", err)
+	}
+	if !redraw {
+		t.Fatal("an expose event must request a redraw")
+	}
+}
+
+// TestResolveLivenessWaitsForReplyOnReplyBearingCookie is the behavioral
+// guard for the hardware-proven root cause. GetInputFocus issues a
+// reply-bearing cookie (NewCookie(true, true)); resolveLiveness must
+// consume it with Reply — meaning it blocks awaiting the server's reply —
+// never with Check, which would return the XGB reply-cookie error on every
+// call regardless of server health. A reply-bearing cookie can be built
+// without a server because xgb's NewCookie does not touch the connection.
+func TestResolveLivenessWaitsForReplyOnReplyBearingCookie(t *testing.T) {
+	conn := &xgb.Conn{}
+	cookie := xproto.GetInputFocusCookie{Cookie: conn.NewCookie(true, true)}
+
+	done := make(chan error, 1)
+	go func() { done <- resolveLiveness(cookie) }()
+	select {
+	case err := <-done:
+		t.Fatalf("resolveLiveness returned %v without any server reply; the cookie was not resolved with Reply", err)
+	case <-time.After(100 * time.Millisecond):
+		// Still waiting for the reply: Reply-resolution semantics intact.
+	}
+}
+
+// TestCheckOnReplyBearingCookieIsTheDeviceRootCause pins the exact xgb
+// behavior the fix depends on, using the real cookie API: Check on a
+// reply-bearing cookie returns the device-proven error immediately, while
+// Reply is the only resolution that waits for the server. This is the
+// behavioral contrast proving connLivenessProbe/resolveLiveness cannot
+// reintroduce the open_failed reconnect loop.
+func TestCheckOnReplyBearingCookieIsTheDeviceRootCause(t *testing.T) {
+	conn := &xgb.Conn{}
+	cookie := xproto.GetInputFocusCookie{Cookie: conn.NewCookie(true, true)}
+
+	err := cookie.Check()
+	if err == nil {
+		t.Fatal("Check on a reply-bearing cookie must fail; the device root cause would be reintroduced")
+	}
+	const deviceError = "Cannot call 'Check' on a cookie that is expecting a *reply*. Use 'Reply' instead."
+	if err.Error() != deviceError {
+		t.Fatalf("Check error=%q want the device-proven %q", err.Error(), deviceError)
+	}
+}

internal/display/x11.go

@@ -178,6 +178,44 @@ func (o X11Opener) pollInterval() time.Duration {
 	return 100 * time.Millisecond
 }
 
+// livenessRequest is one liveness round trip on a connection: it resolves
+// its request and reports only the error.
+type livenessRequest func() error
+
+// connLivenessProbe returns the liveness round trip used after mapping a
+// window and while a session waits for events. GetInputFocus carries a
+// reply, so its cookie must be resolved with Reply: calling Check on a
+// reply-bearing cookie fails at runtime with the XGB reply-cookie error
+// ("Cannot call 'Check' on a cookie that is expecting a *reply*"), which
+// killed every layer open and reconnect on the device.
+func connLivenessProbe(conn *xgb.Conn) livenessRequest {
+	return func() error {
+		return resolveLiveness(xproto.GetInputFocus(conn))
+	}
+}
+
+// resolveLiveness consumes one issued reply-bearing liveness cookie: the
+// reply is read and discarded, and only the error propagates. The cookie
+// must be resolved with Reply exactly once — Check on a reply-bearing
+// cookie returns the XGB reply-cookie error instead of ever reaching the
+// server, so a probe resolved with Check fails on every call regardless of
+// server health.
+func resolveLiveness(cookie xproto.GetInputFocusCookie) error {
+	_, err := cookie.Reply()
+	return err
+}
+
+// confirmMapped runs the post-map liveness stage of Open: a successful
+// reply proves the window is mapped and the server alive; a failure is
+// wrapped as the map-kiosk-window error so the caller closes the
+// connection and drops the mapped window.
+func confirmMapped(probe livenessRequest) error {
+	if err := probe(); err != nil {
+		return fmt.Errorf("map kiosk window: %w", err)
+	}
+	return nil
+}
+
 // Open connects to the X server and maps the kiosk window. Failure is
 // expected while the display manager is starting; the caller retries.
 func (o X11Opener) Open(ctx context.Context) (Session, error) {
@@ -239,10 +277,12 @@ func (o X11Opener) Open(ctx context.Context) (Session, error) {
 	// destroy is video end. WFB and launcher processes are never consulted.
 	xproto.ChangeWindowAttributes(conn, screen.Root,
 		xproto.CwEventMask, []uint32{xproto.EventMaskSubstructureNotify})
-	// Round trip confirms the window is mapped and the server is alive.
-	if err := xproto.GetInputFocus(conn).Check(); err != nil {
+	// Round trip confirms the window is mapped and the server is alive;
+	// the reply-bearing cookie is resolved with Reply (see
+	// connLivenessProbe).
+	if err := confirmMapped(connLivenessProbe(conn)); err != nil {
 		conn.Close()
-		return nil, fmt.Errorf("map kiosk window: %w", err)
+		return nil, err
 	}
 
 	// A layer reconnect must never cover running video: if video windows
@@ -301,6 +341,8 @@ func (o X11Opener) Open(ctx context.Context) (Session, error) {
 		markerAtom:   markerAtom,
 		role:         o.Role,
 		unicodeFont:  unicodeFont,
+		pollEvent:    conn.PollForEvent,
+		probe:        connLivenessProbe(conn),
 	}, nil
 }
 
@@ -560,6 +602,12 @@ type x11Session struct {
 	markerAtom   xproto.Atom
 	role         LayerRole
 	unicodeFont  bool
+	// pollEvent drains one pending X event; conn.PollForEvent in
+	// production. Tests script it.
+	pollEvent func() (xgb.Event, xgb.Error)
+	// probe performs one reply-bearing liveness round trip; its error ends
+	// the session. connLivenessProbe in production. Tests script it.
+	probe livenessRequest
 }
 
 func (s *x11Session) Canvas() Canvas { return s }
@@ -598,7 +646,7 @@ func (s *x11Session) WaitEvent(ctx context.Context) (bool, error) {
 	defer ticker.Stop()
 	for {
 		for {
-			event, err := s.conn.PollForEvent()
+			event, err := s.pollEvent()
 			if err != nil {
 				return false, err
 			}
@@ -613,8 +661,9 @@ func (s *x11Session) WaitEvent(ctx context.Context) (bool, error) {
 			}
 		}
 		// Round trip keeps the connection liveness visible even without
-		// incoming events.
-		if err := xproto.GetInputFocus(s.conn).Check(); err != nil {
+		// incoming events; the reply-bearing cookie is resolved with Reply
+		// (see connLivenessProbe).
+		if err := s.probe(); err != nil {
 			return false, err
 		}
 		select {