fix(agent): capture ScaleToZero wake frames in the runtime host netns

internal/agent/scaletozero_linux.go

@@ -134,10 +134,54 @@ func NewLinuxScaleToZero(bufSize int, interval time.Duration) *ScaleToZero {
 	return newScaleToZero(netlinkLinkStats, afpacketSource{}, interval, bufSize)
 }
 
-// NewLinuxScaleToZeroWithRuntime wires TAP statistics through the node runtime,
-// which owns the TAP network namespace, while retaining the packet capture hook.
+// NewLinuxScaleToZeroWithRuntime wires TAP statistics AND wake-on-traffic
+// packet capture through the node runtime, which owns the host network
+// namespace where the impbr-* bridges and VM TAPs live. The agent's own
+// namespace cannot observe bridged overlay traffic: an AF_PACKET socket opened
+// inside the agent pod only ever sees pod-side frames (loopback, pod veths,
+// Service IPs), never the overlay frames a suspended VM needs to be woken by
+// (validated by the nightly datapath E2E failure streak starting 2026-08-29,
+// where framesSeen incremented but lastDstIP was always pod traffic). The
+// activator polls the runtime at interval/2 so a wake frame registers well
+// inside a 2-minute resume window.
 func NewLinuxScaleToZeroWithRuntime(client *runtimeapi.Client, bufSize int, interval time.Duration) *ScaleToZero {
+	poll := interval / 2
+	if poll <= 0 {
+		poll = time.Second
+	}
 	return newScaleToZero(func(iface string) (uint64, error) {
 		return client.LinkStats(context.Background(), iface)
-	}, afpacketSource{}, interval, bufSize)
+	}, runtimeWakeSource{client: client, poll: poll}, interval, bufSize)
+}
+
+// runtimeWakeSource polls the node runtime for destination IPs observed on the
+// host since the previous poll and feeds each to the wake registry.
+type runtimeWakeSource struct {
+	client *runtimeapi.Client
+	poll   time.Duration
+}
+
+func (r runtimeWakeSource) Run(ctx context.Context, onDstIP func(string)) error {
+	logf.Log.Info("runtime wake source: polling host wake hits via runtime RPC",
+		"pollInterval", r.poll.String())
+	ticker := time.NewTicker(r.poll)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+		}
+		ips, err := r.client.WakeHits(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+			logf.Log.Error(err, "runtime wake source: poll failed; retrying next interval")
+			continue
+		}
+		for _, ip := range ips {
+			onDstIP(ip)
+		}
+	}
 }

internal/agent/scaletozero_wake_runtime_linux_test.go

@@ -0,0 +1,93 @@
+//go:build linux
+
+package agent
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	impdevv1alpha1 "github.com/syscode-labs/imp/api/v1alpha1"
+	"github.com/syscode-labs/imp/internal/runtimeapi"
+)
+
+func TestScaleToZeroRuntimeWakeSourceWakesOnRuntimeObservedFrame(t *testing.T) {
+	t.Parallel()
+
+	const overlayIP = "10.45.0.3"
+	hits := [][]string{
+		{"10.244.0.8", "10.96.0.1"},         // poll 1: pod traffic only — no match
+		{overlayIP, "127.0.0.1", overlayIP}, // poll 2: the wake frame arrives
+	}
+	poll := 0
+	server := runtimeapi.NewServer(runtimeapi.BackendFuncs{
+		LinkStatsFunc: func(_ context.Context, _ string) (uint64, error) {
+			return 42, nil // constant: observe() never reports idle
+		},
+		WakeHitsFunc: func(_ context.Context) ([]string, error) {
+			if poll >= len(hits) {
+				return nil, nil
+			}
+			got := hits[poll]
+			poll++
+			return got, nil
+		},
+	})
+	dir, err := os.MkdirTemp("/tmp", "imp-runtime-")
+	if err != nil {
+		t.Fatalf("MkdirTemp() error = %v", err)
+	}
+	t.Cleanup(func() { _ = os.RemoveAll(dir) })
+	endpoint := filepath.Join(dir, "runtime.sock")
+	if err := server.Start(endpoint); err != nil {
+		t.Fatalf("Start() error = %v", err)
+	}
+	t.Cleanup(func() { _ = server.Close() })
+
+	sz := NewLinuxScaleToZeroWithRuntime(runtimeapi.NewClient(endpoint), 1, time.Second)
+	key := types.NamespacedName{Namespace: "default", Name: "target"}
+
+	// Register the VM as suspended+awaiting wake, exactly as the reconciler
+	// does when it parks a ScaleToZero VM.
+	sz.reg.register(overlayIP, &impdevv1alpha1.ImpVM{
+		ObjectMeta: metav1.ObjectMeta{Namespace: key.Namespace, Name: key.Name},
+	})
+
+	// Feed the first poll through the wake registry directly: pod traffic
+	// must not signal the suspended VM.
+	_, _, err = sz.observe(key, "imptap-x", time.Minute, time.Now())
+	if err != nil {
+		t.Fatalf("observe() error = %v", err)
+	}
+	// Drain hits via the same path the activator uses.
+	src := runtimeWakeSource{client: runtimeapi.NewClient(endpoint), poll: 50 * time.Millisecond}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		_ = src.Run(ctx, sz.reg.onDstIP)
+		close(done)
+	}()
+
+	// Wait until the second poll (which contains the overlay IP) has been
+	// consumed — the wake registry must have signalled the VM.
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		sz.reg.mu.Lock()
+		signalled := sz.reg.signalled[key]
+		sz.reg.mu.Unlock()
+		if signalled {
+			cancel()
+			<-done
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	cancel()
+	<-done
+	t.Fatal("wake registry never signalled the VM despite the runtime observing a frame for its overlay IP")
+}

internal/noderuntime/backend.go

@@ -29,6 +29,12 @@ type Backend struct {
 	Net    network.NetManager
 	// LinkStatsFunc is an optional test seam; production uses runtime-owned netlink.
 	LinkStatsFunc func(context.Context, string) (uint64, error)
+	// WakeHitsFunc is an optional test seam; production uses a host-netns
+	// AF_PACKET capture started lazily on first poll.
+	WakeHitsFunc func(context.Context) ([]string, error)
+	// wakeHitsState holds platform-specific capture state (linux only; nil
+	// elsewhere). Managed entirely by the platform WakeHits implementation.
+	wakeHitsState any
 	// StatePath stores VM inventory that survives a runtime process restart.
 	StatePath string
 }

internal/noderuntime/wakehits_linux.go

@@ -0,0 +1,121 @@
+//go:build linux
+
+package noderuntime
+
+import (
+	"context"
+	"net"
+	"sync"
+	"time"
+
+	"golang.org/x/sys/unix"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+// wakeHits records destination IPs of inbound IPv4 frames observed in the
+// runtime's network namespace (the host netns, where the impbr-* bridges and
+// VM TAPs live). The agent polls WakeHits and wakes any suspended ScaleToZero
+// VM whose overlay IP appears in the drained set.
+type wakeHits struct {
+	mu   sync.Mutex
+	hits map[string]bool
+}
+
+func (w *wakeHits) add(ip string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.hits[ip] = true
+}
+
+func (w *wakeHits) take() []string {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	ips := make([]string, 0, len(w.hits))
+	for ip := range w.hits {
+		ips = append(ips, ip)
+	}
+	w.hits = map[string]bool{}
+	return ips
+}
+
+func htons16(v uint16) uint16 { return v<<8 | v>>8 }
+
+// startHostWakeCapture opens one AF_PACKET socket bound to ETH_P_ALL in the
+// runtime's network namespace and records each inbound IPv4 frame's
+// destination IP into hits until ctx is cancelled.
+//
+// The socket MUST use ETH_P_ALL rather than ETH_P_IP: a specific EtherType
+// registers in the kernel's ptype_base hash, which a bridged device's
+// rx_handler (br_handle_frame, RX_HANDLER_CONSUMED) never reaches. Only
+// ptype_all (protocol 0) is consulted before the rx_handler, so it is the only
+// way to see frames switched onto an impbr-* bridge. IPv4 filtering therefore
+// happens in userspace. Do not "simplify" this back to ETH_P_IP — it silently
+// stops seeing all bridged VM traffic (see the longer analysis in
+// internal/agent/scaletozero_linux.go and imp#35).
+func startHostWakeCapture(ctx context.Context, hits *wakeHits) {
+	log := logf.Log.WithName("wake-capture")
+	fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, int(htons16(unix.ETH_P_ALL)))
+	if err != nil {
+		log.Error(err, "wake-capture: failed to open AF_PACKET socket in host netns; wake-on-traffic is unavailable on this node")
+		return
+	}
+	log.Info("wake-capture: AF_PACKET socket open in host netns, capturing inbound IPv4 frames")
+	go func() {
+		<-ctx.Done()
+		_ = unix.Close(fd)
+	}()
+	buf := make([]byte, 65536)
+	go func() {
+		for {
+			n, _, err := unix.Recvfrom(fd, buf, 0)
+			if err != nil {
+				if ctx.Err() != nil {
+					return
+				}
+				time.Sleep(10 * time.Millisecond)
+				continue
+			}
+			// Frames include the 14-byte Ethernet header; filter to IPv4 via
+			// the EtherType at bytes 12..14 and read the destination IPv4
+			// address at bytes 30..34 (eth[14] + ipv4[16..20]).
+			if n < 34 {
+				continue
+			}
+			if ethType := uint16(buf[12])<<8 | uint16(buf[13]); ethType != 0x0800 {
+				continue
+			}
+			hits.add(net.IP(buf[30:34]).String())
+		}
+	}()
+}
+
+// backendWakeState holds the lazy capture state that only exists on Linux.
+type backendWakeState struct {
+	hits   *wakeHits
+	cancel context.CancelFunc
+}
+
+// wakeStateOf lazily initialises b's wake capture on first poll.
+func wakeStateOf(b *Backend) *backendWakeState {
+	if state, ok := b.wakeHitsState.(*backendWakeState); ok {
+		return state
+	}
+	state := &backendWakeState{hits: &wakeHits{hits: map[string]bool{}}}
+	// The capture intentionally lives for the whole runtime process; it is
+	// never cancelled short of process exit, so the cancel func stays stored
+	// but uncalled.
+	ctx, cancel := context.WithCancel(context.Background()) //nolint:gosec // intentional process-lifetime capture
+	state.cancel = cancel
+	startHostWakeCapture(ctx, state.hits)
+	b.wakeHitsState = state
+	return state
+}
+
+// WakeHits drains the destination IPs observed on the host since the previous
+// call. Implements runtimeapi.Backend.
+func (b *Backend) WakeHits(_ context.Context) ([]string, error) {
+	if b.WakeHitsFunc != nil {
+		return b.WakeHitsFunc(context.Background())
+	}
+	return wakeStateOf(b).hits.take(), nil
+}

internal/noderuntime/wakehits_other.go

@@ -0,0 +1,14 @@
+//go:build !linux
+
+package noderuntime
+
+import (
+	"context"
+	"errors"
+)
+
+// WakeHits is unsupported off Linux; the agent falls back to no-op polling.
+func (b *Backend) WakeHits(_ context.Context) ([]string, error) {
+	_ = b.wakeHitsState // linux-only capture state; unused here
+	return nil, errors.New("runtime wake hits is not supported on this platform")
+}

internal/runtimeapi/runtime.go

@@ -49,6 +49,7 @@ type Backend interface {
 	EnsureVXLAN(context.Context, uint32, string, string, string) error
 	SyncFDB(context.Context, string, []network.FDBEntry) error
 	LinkStats(context.Context, string) (uint64, error)
+	WakeHits(context.Context) ([]string, error)
 }
 
 // BackendFuncs makes protocol behavior testable without Firecracker.
@@ -70,6 +71,7 @@ type BackendFuncs struct {
 	EnsureVXLANFunc      func(context.Context, uint32, string, string, string) error
 	SyncFDBFunc          func(context.Context, string, []network.FDBEntry) error
 	LinkStatsFunc        func(context.Context, string) (uint64, error)
+	WakeHitsFunc         func(context.Context) ([]string, error)
 }
 
 func (f BackendFuncs) Start(ctx context.Context, vm *impdevv1alpha1.ImpVM) (int64, error) {
@@ -179,6 +181,13 @@ func (f BackendFuncs) LinkStats(ctx context.Context, tapName string) (uint64, er
 	return f.LinkStatsFunc(ctx, tapName)
 }
 
+func (f BackendFuncs) WakeHits(ctx context.Context) ([]string, error) {
+	if f.WakeHitsFunc == nil {
+		return nil, errors.New("wake hits is not supported")
+	}
+	return f.WakeHitsFunc(ctx)
+}
+
 // Server serves one runtime backend over a node-local Unix socket.
 type Server struct {
 	backend  Backend
@@ -252,6 +261,7 @@ type StateReply struct{ State VMState }
 type SnapshotReply struct{ Result SnapshotResult }
 type AliveReply struct{ Alive bool }
 type LinkStatsReply struct{ Bytes uint64 }
+type WakeHitsReply struct{ IPs []string }
 type VSockReply struct {
 	Path  string
 	Found bool
@@ -358,6 +368,12 @@ func (s *rpcService) LinkStats(args TAPArgs, reply *LinkStatsReply) error {
 	return err
 }
 
+func (s *rpcService) WakeHits(_ Empty, reply *WakeHitsReply) error {
+	ips, err := s.backend.WakeHits(context.Background())
+	reply.IPs = ips
+	return err
+}
+
 // Client invokes a node runtime through its Unix socket.
 type Client struct{ endpoint string }
 
@@ -467,3 +483,13 @@ func (c *Client) LinkStats(ctx context.Context, tapName string) (uint64, error)
 	}
 	return reply.Bytes, nil
 }
+
+// WakeHits drains the destination IPs of inbound IPv4 frames observed in the
+// runtime's host network namespace since the previous call.
+func (c *Client) WakeHits(ctx context.Context) ([]string, error) {
+	var reply WakeHitsReply
+	if err := c.call(ctx, "WakeHits", Empty{}, &reply); err != nil {
+		return nil, err
+	}
+	return reply.IPs, nil
+}