randomizedcoder · randomizedcoder · Aug 3, 2026 · Aug 2, 2026
diff --git a/nix/default.nix b/nix/default.nix
@@ -318,6 +318,9 @@ in
       microvm-x86_64-s3parquet-pipeline = microvms.vmsS3Parquet.x86_64;
       microvm-x86_64-valkey = microvms.vmsValkey.x86_64;
       microvm-x86_64-tcp-sink = microvms.vmsTcpSink.x86_64;
+      microvm-x86_64-udp-sink = microvms.vmsUdpSink.x86_64;
+      microvm-x86_64-unix-sink = microvms.vmsUnixSink.x86_64;
+      microvm-x86_64-unixgram-sink = microvms.vmsUnixgramSink.x86_64;
       microvm-x86_64-nats = microvms.vmsNats.x86_64;
       microvm-x86_64-nsq = microvms.vmsNsq.x86_64;
       microvm-x86_64-s3parquet-long = microvms.vmsS3ParquetLong.x86_64;
@@ -340,6 +343,9 @@ in
       test-microvm-lifecycle-x86_64-clickhouse-http = microvms.lifecycleClickHttp.x86_64.fullTest;
       test-microvm-lifecycle-x86_64-valkey = microvms.lifecycleValkey.x86_64.fullTest;
       test-microvm-lifecycle-x86_64-tcp-sink = microvms.lifecycleTcpSink.x86_64.fullTest;
+      test-microvm-lifecycle-x86_64-udp-sink = microvms.lifecycleUdpSink.x86_64.fullTest;
+      test-microvm-lifecycle-x86_64-unix-sink = microvms.lifecycleUnixSink.x86_64.fullTest;
+      test-microvm-lifecycle-x86_64-unixgram-sink = microvms.lifecycleUnixgramSink.x86_64.fullTest;
       test-microvm-lifecycle-x86_64-nats = microvms.lifecycleNats.x86_64.fullTest;
       test-microvm-lifecycle-x86_64-nsq = microvms.lifecycleNsq.x86_64.fullTest;
       test-microvm-lifecycle-x86_64-coverage = microvms.lifecycleCoverage.x86_64.fullTest;
@@ -398,6 +404,18 @@ in
       type = "app";
       program = "${microvms.lifecycleTcpSink.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-tcp-sink";
     };
+    microvm-x86_64-lifecycle-udp-sink = {
+      type = "app";
+      program = "${microvms.lifecycleUdpSink.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-udp-sink";
+    };
+    microvm-x86_64-lifecycle-unix-sink = {
+      type = "app";
+      program = "${microvms.lifecycleUnixSink.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-unix-sink";
+    };
+    microvm-x86_64-lifecycle-unixgram-sink = {
+      type = "app";
+      program = "${microvms.lifecycleUnixgramSink.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-unixgram-sink";
+    };
     microvm-x86_64-lifecycle-nats = {
       type = "app";
       program = "${microvms.lifecycleNats.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-nats";

diff --git a/nix/microvms/default.nix b/nix/microvms/default.nix
@@ -230,11 +230,12 @@ let
       sink = "valkey";
     };
 
-  # tcp-sink lifecycle flavor: xtcp2 streams jsonl over the raw TCP dest to an
-  # in-VM ncat receiver; the self-test validates the received records +
-  # destTCP counter (RAW_SOCKET). Native, no docker/broker.
-  mkOneTcpSink =
-    arch:
+  # socket-sink lifecycle flavors: xtcp2 streams jsonl over the raw
+  # tcp/udp/unix/unixgram dest to an in-VM ncat receiver; the self-test
+  # validates the received records + the per-scheme send counter (RAW_SOCKET).
+  # Native, no docker/broker.
+  mkOneSocketSink =
+    sinkName: arch:
     import ./mkVm.nix {
       inherit
         pkgs
@@ -245,8 +246,12 @@ let
         xtcp2Package
         xtcp2AllPackage
         ;
-      sink = "tcp-sink";
+      sink = sinkName;
     };
+  mkOneTcpSink = mkOneSocketSink "tcp-sink";
+  mkOneUdpSink = mkOneSocketSink "udp-sink";
+  mkOneUnixSink = mkOneSocketSink "unix-sink";
+  mkOneUnixgramSink = mkOneSocketSink "unixgram-sink";
 
   # nats lifecycle flavor: native in-VM NATS server + pre-subscribed consumer.
   mkOneNats =
@@ -413,6 +418,9 @@ let
   vmsValkey = lib.genAttrs constants.supportedArchs mkOneValkey;
 
   vmsTcpSink = lib.genAttrs constants.supportedArchs mkOneTcpSink;
+  vmsUdpSink = lib.genAttrs constants.supportedArchs mkOneUdpSink;
+  vmsUnixSink = lib.genAttrs constants.supportedArchs mkOneUnixSink;
+  vmsUnixgramSink = lib.genAttrs constants.supportedArchs mkOneUnixgramSink;
 
   vmsNats = lib.genAttrs constants.supportedArchs mkOneNats;
 
@@ -444,17 +452,24 @@ let
     };
   });
 
-  lifecycleTcpSink = lib.genAttrs constants.supportedArchs (arch: {
-    fullTest = microvmLib.mkLifecycleFullTest {
-      inherit arch;
-      vm = vmsTcpSink.${arch};
-      suffix = "-tcp-sink";
-      # Baseline sentinels plus the raw-socket verdict. Native (no docker), but
-      # the self-test waits for records to stream over TCP + reach the sink.
-      sentinelRe = "SYSTEMD|METRICS|NETLINK|BINARIES_HELP|GRPC_ROUNDTRIP|POLL_STREAM|HEALTH|OUTPUT_CONTENT|LISTEN_STREAM|RAW_SOCKET|NS_INSPECT|NSTEST|NS_LIFECYCLE|NS_TRAFFIC|NS_DOCKER|NS_ANONYMOUS|CTL_HOT|CTL_TRIGGER|CTL_RESTART|OVERALL";
-      timeoutSec = 240;
-    };
-  });
+  # Lifecycle runner for a socket-sink flavor (tcp/udp/unix/unixgram). Baseline
+  # sentinels plus the raw-socket verdict. Native (no docker), but the self-test
+  # waits for records to stream over the socket + reach the sink.
+  mkLifecycleSocketSink =
+    vmsAttr: suffixName:
+    lib.genAttrs constants.supportedArchs (arch: {
+      fullTest = microvmLib.mkLifecycleFullTest {
+        inherit arch;
+        vm = vmsAttr.${arch};
+        suffix = suffixName;
+        sentinelRe = "SYSTEMD|METRICS|NETLINK|BINARIES_HELP|GRPC_ROUNDTRIP|POLL_STREAM|HEALTH|OUTPUT_CONTENT|LISTEN_STREAM|RAW_SOCKET|NS_INSPECT|NSTEST|NS_LIFECYCLE|NS_TRAFFIC|NS_DOCKER|NS_ANONYMOUS|CTL_HOT|CTL_TRIGGER|CTL_RESTART|OVERALL";
+        timeoutSec = 240;
+      };
+    });
+  lifecycleTcpSink = mkLifecycleSocketSink vmsTcpSink "-tcp-sink";
+  lifecycleUdpSink = mkLifecycleSocketSink vmsUdpSink "-udp-sink";
+  lifecycleUnixSink = mkLifecycleSocketSink vmsUnixSink "-unix-sink";
+  lifecycleUnixgramSink = mkLifecycleSocketSink vmsUnixgramSink "-unixgram-sink";
 
   lifecycleNats = lib.genAttrs constants.supportedArchs (arch: {
     fullTest = microvmLib.mkLifecycleFullTest {
@@ -648,6 +663,9 @@ in
     vmsS3Parquet
     vmsValkey
     vmsTcpSink
+    vmsUdpSink
+    vmsUnixSink
+    vmsUnixgramSink
     vmsNats
     vmsNsq
     vmsS3ParquetLong
@@ -666,6 +684,9 @@ in
     lifecycleS3Parquet
     lifecycleValkey
     lifecycleTcpSink
+    lifecycleUdpSink
+    lifecycleUnixSink
+    lifecycleUnixgramSink
     lifecycleNats
     lifecycleNsq
     lifecycleCoverage

diff --git a/nix/microvms/mkVm.nix b/nix/microvms/mkVm.nix
@@ -49,9 +49,61 @@ let
   # destination end-to-end: xtcp2 streams jsonl records over TCP to an in-VM
   # ncat receiver (dual-stack), and the self-test validates the received
   # records + the destTCP send counter. No docker, no broker.
-  isTcpSink = sink == "tcp-sink";
-  tcpSinkPort = 13001;
-  tcpSinkFile = "/tmp/xtcp2-tcp-sink.out";
+  # socket-sink flavors: <scheme>-sink proves the raw tcp/udp/unix/unixgram
+  # destination end-to-end. xtcp2 streams jsonl records over the dest to a
+  # dual-stack ncat receiver, which writes everything to socketSinkFile for the
+  # RAW_SOCKET self-test check to validate. No docker, no broker.
+  socketSinkScheme =
+    if sink == "tcp-sink" then
+      "tcp"
+    else if sink == "udp-sink" then
+      "udp"
+    else if sink == "unix-sink" then
+      "unix"
+    else if sink == "unixgram-sink" then
+      "unixgram"
+    else
+      "";
+  isSocketSink = socketSinkScheme != "";
+  socketSinkPort = 13001; # inet dests (tcp/udp)
+  socketSinkPath = "/run/xtcp2-sink.sock"; # unix dests
+  socketSinkFile = "/tmp/xtcp2-socket-sink.out";
+  # -dest value per scheme. `localhost` (dual-stack) for inet dests.
+  socketSinkDest =
+    if socketSinkScheme == "tcp" then
+      "tcp:localhost:${toString socketSinkPort}"
+    else if socketSinkScheme == "udp" then
+      "udp:localhost:${toString socketSinkPort}"
+    else
+      "${socketSinkScheme}:${socketSinkPath}"; # unix:/path or unixgram:/path
+  # ncat receiver invocation per scheme (its stdout is redirected to
+  # socketSinkFile by the service). inet dests listen dual-stack on ::
+  # (bindv6only=0 also accepts IPv4-mapped); unix dests bind the socket path.
+  socketSinkReceiverCmd =
+    if socketSinkScheme == "tcp" then
+      "${pkgs.nmap}/bin/ncat --listen --keep-open :: ${toString socketSinkPort}"
+    else if socketSinkScheme == "udp" then
+      "${pkgs.nmap}/bin/ncat --udp --listen --keep-open :: ${toString socketSinkPort}"
+    else if socketSinkScheme == "unix" then
+      "${pkgs.nmap}/bin/ncat --unixsock --listen --keep-open ${socketSinkPath}"
+    else
+      # unixgram: ncat can't LISTEN on a unix DATAGRAM socket (`--unixsock
+      # --udp --listen` fails with "connect: Invalid argument"), so use socat's
+      # UNIX-RECV, which binds a unix datagram socket and streams received
+      # datagrams to stdout (-u = unidirectional recv→stdout).
+      "${pkgs.socat}/bin/socat -u UNIX-RECV:${socketSinkPath} -";
+  # Send-side Writes counter (function, variable) per scheme. udp is labelled
+  # under the legacy Inetdiager/udpWrites; the rest are destXxx/Writes.
+  socketSinkMetricFn =
+    if socketSinkScheme == "udp" then
+      "Inetdiager"
+    else if socketSinkScheme == "tcp" then
+      "destTCP"
+    else if socketSinkScheme == "unix" then
+      "destUnix"
+    else
+      "destUnixGram";
+  socketSinkMetricVar = if socketSinkScheme == "udp" then "udpWrites" else "Writes";
   # minimal = the lifecycle correctness gate. Unlike soak (which shares the
   # basic `-dest null` args), minimal writes jsonl to a file so the self-test
   # can validate the daemon's serialized output content (OUTPUT_CONTENT check).
@@ -199,8 +251,13 @@ let
     runFileOutputCheck = isMinimal;
     inherit fileOutputPath;
     # tcp-sink flavor only: validate records received over the raw TCP dest.
-    runRawSocketCheck = isTcpSink;
-    inherit tcpSinkFile;
+    runRawSocketCheck = isSocketSink;
+    inherit
+      socketSinkScheme
+      socketSinkFile
+      socketSinkMetricFn
+      socketSinkMetricVar
+      ;
   };
 
   # Default monitor cadence for the s3parquet-long flavor. 60 s is fast
@@ -1376,14 +1433,13 @@ let
   # the self-test run as root, so a 0600 file under /var/log is readable.
   fileOutputPath = "/var/log/xtcp2.jsonl";
 
-  # tcp-sink flavor: stream jsonl records over the raw TCP destination to the
-  # in-VM ncat receiver. `localhost` (dual-stack) resolves to 127.0.0.1 AND
-  # ::1; the sink listens on [::] (dual-stack), so whichever family Go's
-  # resolver picks reaches it. jsonl is newline-delimited so the receiver can
-  # count/validate records.
-  xtcp2TcpSinkArgs = [
+  # socket-sink flavors: stream jsonl records over the raw socket destination
+  # (tcp/udp/unix/unixgram) to the in-VM ncat receiver. For inet dests,
+  # `localhost` (dual-stack) resolves to 127.0.0.1 AND ::1 and the sink listens
+  # on [::], so whichever family Go's resolver picks reaches it.
+  xtcp2SocketSinkArgs = [
     "-dest"
-    "tcp:localhost:${toString tcpSinkPort}"
+    socketSinkDest
     "-marshal"
     "jsonl"
     "-frequency"
@@ -2014,9 +2070,10 @@ in
               # minimal (lifecycle) writes jsonl to a file so OUTPUT_CONTENT
               # can validate the daemon's serialized output.
               xtcp2FileArgs
-            else if isTcpSink then
-              # tcp-sink: stream jsonl over the raw TCP dest to the ncat sink.
-              xtcp2TcpSinkArgs
+            else if isSocketSink then
+              # socket-sink: stream jsonl over the raw tcp/udp/unix/unixgram
+              # dest to the ncat sink.
+              xtcp2SocketSinkArgs
             else
               # Soak reuses the basic args (`-dest null`, fast frequency).
               # The point of soak is namespace + netlink churn, not
@@ -2096,21 +2153,23 @@ in
           };
         };
 
-        # tcp-sink flavor: a dual-stack ncat receiver that appends everything
-        # xtcp2 streams over the raw TCP destination to tcpSinkFile, for the
-        # RAW_SOCKET self-test check to count + validate. Listens on [::]
-        # (bindv6only=0 → also accepts IPv4-mapped) so `localhost` on the daemon
-        # side reaches it on either family. Ordered before xtcp2 so the dest
-        # dial doesn't race a missing listener (xtcp2's Restart=on-failure also
-        # covers the race).
-        systemd.services.xtcp2-tcp-sink = lib.mkIf isTcpSink {
-          description = "xtcp2 tcp-sink — dual-stack ncat receiver for the raw TCP dest";
+        # socket-sink flavors: an ncat receiver that appends everything xtcp2
+        # streams over the raw socket destination (tcp/udp/unix/unixgram) to
+        # socketSinkFile, for the RAW_SOCKET self-test check to count +
+        # validate. inet dests listen dual-stack on [::] (bindv6only=0 → also
+        # accepts IPv4-mapped) so `localhost` on the daemon side reaches it on
+        # either family. Ordered before xtcp2 so the dest dial doesn't race a
+        # missing listener (xtcp2's Restart=on-failure also covers the race).
+        systemd.services.xtcp2-socket-sink = lib.mkIf isSocketSink {
+          description = "xtcp2 socket-sink — ncat receiver for the raw ${socketSinkScheme} dest";
           before = [ "xtcp2.service" ];
           wantedBy = [ "multi-user.target" ];
           serviceConfig = {
             Type = "simple";
-            ExecStart = "${pkgs.writeShellScript "xtcp2-tcp-sink" ''
-              exec ${pkgs.nmap}/bin/ncat --listen --keep-open :: ${toString tcpSinkPort} > ${tcpSinkFile}
+            ExecStart = "${pkgs.writeShellScript "xtcp2-socket-sink" ''
+              # A stale unix socket path would make ncat fail to bind on restart.
+              rm -f ${socketSinkPath} 2>/dev/null || true
+              exec ${socketSinkReceiverCmd} > ${socketSinkFile}
             ''}";
             Restart = "on-failure";
             RestartSec = "1s";

diff --git a/nix/microvms/self-test.nix b/nix/microvms/self-test.nix
@@ -141,11 +141,15 @@
   # wouldn't exist — hence the gate.
   runFileOutputCheck ? false,
   fileOutputPath ? "/var/log/xtcp2.jsonl",
-  # When true (tcp-sink flavor), the RAW_SOCKET check validates the records
-  # xtcp2 streamed over the raw TCP destination to the in-VM ncat receiver
-  # (written to tcpSinkFile).
+  # When true (socket-sink flavors), the RAW_SOCKET check validates the records
+  # xtcp2 streamed over the raw socket destination (tcp/udp/unix/unixgram) to
+  # the in-VM ncat receiver (written to socketSinkFile), and the per-scheme
+  # send-side Writes counter (socketSinkMetricFn/Var).
   runRawSocketCheck ? false,
-  tcpSinkFile ? "/tmp/xtcp2-tcp-sink.out",
+  socketSinkScheme ? "tcp",
+  socketSinkFile ? "/tmp/xtcp2-socket-sink.out",
+  socketSinkMetricFn ? "destTCP",
+  socketSinkMetricVar ? "Writes",
 }:
 
 pkgs.writeShellApplication {
@@ -480,45 +484,49 @@ pkgs.writeShellApplication {
     fi
     if [ "$check5e" -ne 0 ]; then overall_ok=0; fi
 
-    # ─── Check 5f: raw TCP destination → in-VM ncat sink ──────────────────
-    # (tcp-sink flavor only). xtcp2 streams jsonl records over `-dest tcp:...`
-    # to a dual-stack ncat receiver; validate the received records ARE
-    # well-formed jsonl (send + receipt) AND the destTCP write counter grew.
-    # tcp/udp/unix dests were Go-component-only before this.
+    # ─── Check 5f: raw socket destination → in-VM ncat sink ───────────────
+    # (socket-sink flavors: tcp/udp/unix/unixgram). xtcp2 streams jsonl records
+    # over `-dest <scheme>:...` to an ncat receiver; validate the received
+    # records carry a non-empty hostname (send + receipt) AND the per-scheme
+    # send counter ran. These dests were Go-component-only before this.
     ${lib.optionalString runRawSocketCheck ''
-      echo "--- check 5f: raw TCP dest → ncat sink (${tcpSinkFile}) ---"
+      echo "--- check 5f: raw ${socketSinkScheme} dest → ncat sink (${socketSinkFile}) ---"
       check5f=1
       # Wait for records to arrive at the sink file.
       for _ in $(seq 1 20); do
-        if [ -s "${tcpSinkFile}" ]; then break; fi
+        if [ -s "${socketSinkFile}" ]; then break; fi
         sleep 1
       done
-      # destTCP Writes counter (>=1 = the daemon's send side ran). Not a growth
-      # check: the daemon has usually already flushed its writes by the time
-      # this check runs, so require the counter to be non-zero, not growing.
-      writes=$(metric_value "xtcp_counts" 'function="destTCP"' 'variable="Writes"')
-      # Snapshot + validate only complete lines (drop a partial trailing line —
-      # ncat may be mid-write). Same wc -l / head -n idiom as OUTPUT_CONTENT.
-      cp "${tcpSinkFile}" /tmp/xtcp2-tcpsink.snap 2>/dev/null || true
-      total=$(wc -l < /tmp/xtcp2-tcpsink.snap 2>/dev/null || echo 0)
+      # Per-scheme Writes counter (>=1 = the daemon's send side ran). Not a
+      # growth check: the daemon has usually already flushed by the time this
+      # runs, so require non-zero, not growing.
+      writes=$(metric_value "xtcp_counts" 'function="${socketSinkMetricFn}"' 'variable="${socketSinkMetricVar}"')
+      # Snapshot, drop a possibly-partial trailing line (ncat mid-write), then
+      # extract JSON objects: `grep '{...}'` strips the unix STREAM dest's
+      # varint length prefix and is a no-op for the newline-delimited jsonl of
+      # tcp/udp/unixgram. jq-validate each carries a non-empty hostname.
+      cp "${socketSinkFile}" /tmp/xtcp2-sink.snap 2>/dev/null || true
+      total=$(wc -l < /tmp/xtcp2-sink.snap 2>/dev/null || echo 0)
       total=''${total:-0}
       good=0
       bad=0
-      while IFS= read -r line; do
-        [ -z "$line" ] && continue
-        if printf '%s\n' "$line" \
+      while IFS= read -r obj; do
+        [ -z "$obj" ] && continue
+        if printf '%s\n' "$obj" \
           | jq -e 'has("hostname") and (.hostname | length > 0)' >/dev/null 2>&1; then
           good=$((good + 1))
         else
           bad=$((bad + 1))
         fi
-      done < <(head -n "$total" /tmp/xtcp2-tcpsink.snap 2>/dev/null)
-      if [ "$good" -ge 1 ] && [ "$bad" -eq 0 ] && [ "''${writes:-0}" -ge 1 ] 2>/dev/null; then
-        echo "XTCP2_SELF_TEST_RAW_SOCKET_PASS  (records=$good, destTCP_Writes=$writes)"
+      done < <(head -n "$total" /tmp/xtcp2-sink.snap 2>/dev/null | grep -aoE '\{.*\}')
+      # good>=1 + counter>=1 is the proof; a few `bad` are tolerated (a varint
+      # length byte can occasionally look like '{' on the unix stream framing).
+      if [ "$good" -ge 1 ] && [ "''${writes:-0}" -ge 1 ] 2>/dev/null; then
+        echo "XTCP2_SELF_TEST_RAW_SOCKET_PASS  (scheme=${socketSinkScheme}, records=$good, malformed=$bad, ${socketSinkMetricFn}/${socketSinkMetricVar}=$writes)"
         check5f=0
       else
-        echo "XTCP2_SELF_TEST_RAW_SOCKET_FAIL  (records=$good, malformed=$bad, destTCP_Writes=$writes)"
-        head -n 3 /tmp/xtcp2-tcpsink.snap 2>/dev/null || echo "(no sink file)"
+        echo "XTCP2_SELF_TEST_RAW_SOCKET_FAIL  (scheme=${socketSinkScheme}, records=$good, malformed=$bad, ${socketSinkMetricFn}/${socketSinkMetricVar}=$writes)"
+        head -c 200 /tmp/xtcp2-sink.snap 2>/dev/null || echo "(no sink file)"
       fi
       if [ "$check5f" -ne 0 ]; then overall_ok=0; fi
     ''}