test(inttest): raw TCP destination e2e flavor (tcp-sink)

nix/default.nix

@@ -317,6 +317,7 @@ in
       microvm-x86_64-clickhouse-pipeline-parquet = microvms.vmsClickPipeParquet.x86_64;
       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-nats = microvms.vmsNats.x86_64;
       microvm-x86_64-nsq = microvms.vmsNsq.x86_64;
       microvm-x86_64-s3parquet-long = microvms.vmsS3ParquetLong.x86_64;
@@ -338,6 +339,7 @@ in
       test-microvm-lifecycle-x86_64-s3parquet = microvms.lifecycleS3Parquet.x86_64.fullTest;
       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-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;
@@ -392,6 +394,10 @@ in
       type = "app";
       program = "${microvms.lifecycleValkey.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-valkey";
     };
+    microvm-x86_64-lifecycle-tcp-sink = {
+      type = "app";
+      program = "${microvms.lifecycleTcpSink.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-tcp-sink";
+    };
     microvm-x86_64-lifecycle-nats = {
       type = "app";
       program = "${microvms.lifecycleNats.x86_64.fullTest}/bin/xtcp2-lifecycle-full-test-x86_64-nats";

nix/microvms/default.nix

@@ -230,6 +230,24 @@ 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:
+    import ./mkVm.nix {
+      inherit
+        pkgs
+        lib
+        microvm
+        nixpkgs
+        arch
+        xtcp2Package
+        xtcp2AllPackage
+        ;
+      sink = "tcp-sink";
+    };
+
   # nats lifecycle flavor: native in-VM NATS server + pre-subscribed consumer.
   mkOneNats =
     arch:
@@ -394,6 +412,8 @@ let
 
   vmsValkey = lib.genAttrs constants.supportedArchs mkOneValkey;
 
+  vmsTcpSink = lib.genAttrs constants.supportedArchs mkOneTcpSink;
+
   vmsNats = lib.genAttrs constants.supportedArchs mkOneNats;
 
   vmsNsq = lib.genAttrs constants.supportedArchs mkOneNsq;
@@ -424,6 +444,18 @@ 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;
+    };
+  });
+
   lifecycleNats = lib.genAttrs constants.supportedArchs (arch: {
     fullTest = microvmLib.mkLifecycleFullTest {
       inherit arch;
@@ -615,6 +647,7 @@ in
     vmsClickPipeParquet
     vmsS3Parquet
     vmsValkey
+    vmsTcpSink
     vmsNats
     vmsNsq
     vmsS3ParquetLong
@@ -632,6 +665,7 @@ in
     lifecycleClickHttp
     lifecycleS3Parquet
     lifecycleValkey
+    lifecycleTcpSink
     lifecycleNats
     lifecycleNsq
     lifecycleCoverage

nix/microvms/mkVm.nix

@@ -45,6 +45,13 @@ let
   isCoverage = sink == "coverage" || sink == "coverage-iouring";
   isCoverageIoUring = sink == "coverage-iouring";
   isSoak = sink == "soak";
+  # tcp-sink = a lightweight lifecycle flavor that proves the raw `tcp`
+  # 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";
   # 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).
@@ -191,6 +198,9 @@ let
     # minimal flavor only: validate the daemon's jsonl file-dest output.
     runFileOutputCheck = isMinimal;
     inherit fileOutputPath;
+    # tcp-sink flavor only: validate records received over the raw TCP dest.
+    runRawSocketCheck = isTcpSink;
+    inherit tcpSinkFile;
   };
 
   # Default monitor cadence for the s3parquet-long flavor. 60 s is fast
@@ -1366,6 +1376,22 @@ 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 = [
+    "-dest"
+    "tcp:localhost:${toString tcpSinkPort}"
+    "-marshal"
+    "jsonl"
+    "-frequency"
+    "2s"
+    "-timeout"
+    "1s"
+  ];
+
   # minimal (lifecycle) flavor: same fast cadence as basic, but write records
   # as jsonl to a file so the self-test's OUTPUT_CONTENT check can validate the
   # daemon's serialized output (jsonl marshaller → file destination → record
@@ -1988,6 +2014,9 @@ 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
               # Soak reuses the basic args (`-dest null`, fast frequency).
               # The point of soak is namespace + netlink churn, not
@@ -2067,6 +2096,28 @@ 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";
+          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}
+            ''}";
+            Restart = "on-failure";
+            RestartSec = "1s";
+            StandardError = "journal+console";
+          };
+        };
+
         # Soak flavor: long-running services that churn namespaces + scrape
         # /metrics into a file inside the VM. The host-side soak runner
         # (see nix/microvms/lib.nix mkSoakRunner) boots the VM, sleeps for

nix/microvms/self-test.nix

@@ -141,6 +141,11 @@
   # 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).
+  runRawSocketCheck ? false,
+  tcpSinkFile ? "/tmp/xtcp2-tcp-sink.out",
 }:
 
 pkgs.writeShellApplication {
@@ -475,6 +480,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.
+    ${lib.optionalString runRawSocketCheck ''
+      echo "--- check 5f: raw TCP dest → ncat sink (${tcpSinkFile}) ---"
+      check5f=1
+      # Wait for records to arrive at the sink file.
+      for _ in $(seq 1 20); do
+        if [ -s "${tcpSinkFile}" ]; 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)
+      total=''${total:-0}
+      good=0
+      bad=0
+      while IFS= read -r line; do
+        [ -z "$line" ] && continue
+        if printf '%s\n' "$line" \
+          | 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)"
+        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)"
+      fi
+      if [ "$check5f" -ne 0 ]; then overall_ok=0; fi
+    ''}
+
     # ─── Check 6: ns inspector reads netns state ─────────────────────────
     echo "--- check 6: ns inspector ---"
     check6=1