FP-6: bounded S3 queue that never blocks polling; string policy for tenant/network strings

CHANGELOG.md

@@ -15,6 +15,8 @@
 
 ### New
 
+- **The S3 upload queue never blocks polling.** It used to block the poller once 16 envelopes were waiting for a slow or unreachable S3. Now `-s3QueueDepth` (default 16, max 4096) and `-s3MaxBufferedBytes` (default 64 MiB, `0` = no byte bound) bound it, and past either it drops the oldest envelope. Drops are counted in `xtcp_s3_dropped_total{reason}` and `xtcp_s3_dropped_bytes_total{reason}`; `xtcp_s3_buffered_bytes{stage}` shows what is held, and `queueFull` now counts drops. Env `S3_QUEUE_DEPTH`, `S3_MAX_BUFFERED_BYTES`; startup-only (`ConfigService.Set` keeps the running values). Budget ≈ `-s3MaxBufferedBytes` + the flush threshold + one upload ([design](docs/design-jitter-and-backoff.md#back-pressure--memory)).
+- **Strings from tenants and the network are cleaned.** Container names and images, a tenant's interface names, LLDP and NIC strings, the ASN feed's network owner, netns names and congestion names are cut to 256 bytes (16 for interface and congestion names) at a UTF-8 boundary. Invalid UTF-8 and control characters (C0, DEL, C1) become U+FFFD. Each is cleaned once, where it enters, not per row.
 - **Per-namespace caps.** `-maxDumpBytes` (default 256 MiB) and `-maxRecordsPerNetns` (default 262144) bound one namespace's dump per poll: packets past the byte cap are drained undecoded, and records past the record cap are dropped before decode. `-maxNetns` (default 4096) bounds the namespaces tracked. At the cap, tracked namespaces keep their slots, and new ones wait for a freed slot, first come, first served. Alert on `xtcp_netns_dropped_total`. Drops are counted in `xtcp_dump_dropped_total{reason}` and `xtcp_netns_dropped_total`, with one log line per truncated dump or short reconcile. `0` turns a cap off; the matching env vars are `MAX_DUMP_BYTES`, `MAX_RECORDS_PER_NETNS` and `MAX_NETNS`. The caps are startup-only: `ConfigService.Set` keeps the running values. No healthy host should reach the defaults; a host that does now loses that namespace's excess rows, where it used to collect them all ([polling and batching](docs/polling-and-batching.md#per-namespace-caps)).
 - **Socket-count gate, reconcile limits, per-owner quota.** Before each poll xtcp2 reads each namespace's sockstat (an fd opened in the namespace; `pread`, or io_uring batches with `-ioUring`) and, when the total is over `-maxRecordsPerPoll` (default 1,048,576), degrades only the namespaces over their fair share: first TIME_WAIT/SYN_RECV are not asked for, then one of K port shards is dumped per poll. `-maxNetnsPerOwner` (default 64) caps namespaces tracked per owner (the container runtime's cgroup), `-maxNetnsAddsPerReconcile` (256) the adds per reconcile, and `-reconcileTimeout` (1s) the namespace scan; a timed-out scan deletes nothing. New metrics `xtcp_poll_sockets`, `xtcp_poll_degraded_total{level}` and `xtcp_reconcile_incomplete_total{reason}`; `xtcp_netns_dropped_total` gains `reason`. `0` turns each off; all are startup-only. Each tracked namespace now holds four fds (was two: sockstat and `ip_local_port_range` are new), and xtcp2 warns at startup when `RLIMIT_NOFILE` can't cover them ([polling and batching](docs/polling-and-batching.md#socket-count-gate), [network namespaces](docs/network-namespaces.md#configuration)). The analysis behind an upstream all-namespace dump is in [docs/kernel](docs/kernel/inet-diag-all-netns.md).
 - **Vulnerability check.** `checks.govulncheck` runs govulncheck against a pinned snapshot of the Go vulnerability database, with no tags and with every flavor tag, and fails on a reachable finding not in its accepted list. One is accepted: GO-2026-6859 in `net/http`, fixed in go1.27.4, until the toolchain moves. `nix run .#govulncheck-fresh` checks against the live database and runs `go mod verify`; `nix build .#sbom-<binary>` writes a CycloneDX SBOM for any binary package.

cmd/xtcp2/testdata/defineflags-golden.txt

@@ -120,8 +120,10 @@ s3Endpoint		s3parquet: S3-compatible endpoint URL (e.g. http://127.0.0.1:9000 fo
 s3FlushInterval	0s	s3parquet: staleness ceiling — force-flush the in-memory Parquet object after this long even if under the byte cap. 0 = derive as max(-frequency, 30m).
 s3FlushJitterPct	20	s3parquet: jitter as a percent (0-100) of -s3FlushInterval, applied to the timed flush so the fleet doesn't ceiling-flush in lockstep. 0 disables.
 s3FlushThresholdJitterPct	20	s3parquet: per-object downward jitter as a percent (0-100) of the byte cap; each object finalizes at threshold*(1-rand[0,pct/100]) to de-sync the size-cap upload path. 0 disables.
+s3MaxBufferedBytes	67108864	s3parquet: marshaled envelope bytes the upload queue may hold. Past it the oldest envelope is dropped (reason="buffered_bytes"); the newest is always kept. 0 = no byte bound. Falls back to S3_MAX_BUFFERED_BYTES env.
 s3ParquetFlushBytes	0	s3parquet: soft cap on the in-memory Parquet builder's uncompressed row bytes before finalize+upload. 0 = daemon default (63 MiB).
 s3Prefix		s3parquet: optional key prefix within the bucket. Combined with Hive-style partitioning host=…/date=…/hour=…/<file>.parquet.
+s3QueueDepth	16	s3parquet: envelopes queued for the upload worker (max 4096). Past it the oldest is dropped and counted in xtcp_s3_dropped_total{reason="queue_depth"}, so an S3 outage never blocks polling. 0 = 16. Falls back to S3_QUEUE_DEPTH env.
 s3Region		s3parquet: S3 region. Defaults to 'us-east-1' when empty; required by AWS, ignored by most MinIO setups.
 s3SecretKey		s3parquet: S3 secret key. Prefer S3_SECRET_KEY_FILE (path to a file, mode 0400, holding the key): a flag value shows in ps and docker inspect, warns, and is refused by production builds. Never logged.
 s3SkipBucketProbe	false	s3parquet: skip the startup BucketExists probe (a HeadBucket needing s3:ListBucket). Set true for a write-only, s3:PutObject-only credential. Falls back to S3_SKIP_BUCKET_PROBE env.

cmd/xtcp2/testdata/printflags-golden.txt

@@ -28,6 +28,8 @@
 *s3FlushThresholdJitterPct: 117
 *s3UploadMaxAttempts: 118
 *s3UploadBackoffCap: 119ms
+*s3QueueDepth: 1191
+*s3MaxBufferedBytes: 1192
 *reconcileFrequency: 120ms
 *reconcileBeforePoll: false
 *maxDumpBytes: 1201

cmd/xtcp2/xtcp2.go

@@ -116,12 +116,14 @@ const (
 	// docs/design-jitter-and-backoff.md. A 0 pct disables the corresponding
 	// jitter (restoring deterministic behavior); the two 0-valued durations
 	// mean "derive from poll_frequency" (see the s3parquet destination).
-	pollJitterPctCst             uint = 20
-	s3FlushIntervalCst                = 0 * time.Second
-	s3FlushJitterPctCst          uint = 20
-	s3FlushThresholdJitterPctCst uint = 20
-	s3UploadMaxAttemptsCst       uint = 10
-	s3UploadBackoffCapCst             = 0 * time.Second
+	pollJitterPctCst             uint   = 20
+	s3FlushIntervalCst                  = 0 * time.Second
+	s3FlushJitterPctCst          uint   = 20
+	s3FlushThresholdJitterPctCst uint   = 20
+	s3UploadMaxAttemptsCst       uint   = 10
+	s3UploadBackoffCapCst               = 0 * time.Second
+	s3QueueDepthCst              uint   = 16
+	s3MaxBufferedBytesCst        uint64 = 64 * 1024 * 1024
 
 	// Namespace-reconcile cadence defaults (Method B /proc-scan discovery).
 	// reconcileBeforePoll ties discovery to poll cadence (reconcile before each
@@ -309,6 +311,8 @@ type mainFlags struct {
 	s3FlushThresholdJitterPct *uint
 	s3UploadMaxAttempts       *uint
 	s3UploadBackoffCap        *time.Duration
+	s3QueueDepth              *uint
+	s3MaxBufferedBytes        *uint64
 
 	reconcileFrequency  *time.Duration
 	reconcileBeforePoll *bool
@@ -550,6 +554,8 @@ func defineS3Flags(f *mainFlags) {
 	f.s3FlushThresholdJitterPct = flag.Uint("s3FlushThresholdJitterPct", s3FlushThresholdJitterPctCst, "s3parquet: per-object downward jitter as a percent (0-100) of the byte cap; each object finalizes at threshold*(1-rand[0,pct/100]) to de-sync the size-cap upload path. 0 disables.")
 	f.s3UploadMaxAttempts = flag.Uint("s3UploadMaxAttempts", s3UploadMaxAttemptsCst, "s3parquet: max upload attempts (original + retries) before dropping the object. Retries use full-jitter exponential backoff.")
 	f.s3UploadBackoffCap = flag.Duration("s3UploadBackoffCap", s3UploadBackoffCapCst, "s3parquet: cap on a single upload retry's backoff window (full jitter draws in [0,window]). 0 = derive as clamp(-frequency/10, 1s, 1h).")
+	f.s3QueueDepth = flag.Uint("s3QueueDepth", s3QueueDepthCst, "s3parquet: envelopes queued for the upload worker (max 4096). Past it the oldest is dropped and counted in xtcp_s3_dropped_total{reason=\"queue_depth\"}, so an S3 outage never blocks polling. 0 = 16. Falls back to S3_QUEUE_DEPTH env.")
+	f.s3MaxBufferedBytes = flag.Uint64("s3MaxBufferedBytes", s3MaxBufferedBytesCst, "s3parquet: marshaled envelope bytes the upload queue may hold. Past it the oldest envelope is dropped (reason=\"buffered_bytes\"); the newest is always kept. 0 = no byte bound. Falls back to S3_MAX_BUFFERED_BYTES env.")
 }
 
 // defineCapFlags registers the three caps on what a tenant can cost the
@@ -742,6 +748,8 @@ func printS3Flags(f *mainFlags) {
 	fmt.Println("*s3FlushThresholdJitterPct:", *f.s3FlushThresholdJitterPct)
 	fmt.Println("*s3UploadMaxAttempts:", *f.s3UploadMaxAttempts)
 	fmt.Println("*s3UploadBackoffCap:", *f.s3UploadBackoffCap)
+	fmt.Println("*s3QueueDepth:", *f.s3QueueDepth)
+	fmt.Println("*s3MaxBufferedBytes:", *f.s3MaxBufferedBytes)
 	fmt.Println("*reconcileFrequency:", *f.reconcileFrequency)
 	fmt.Println("*reconcileBeforePoll:", *f.reconcileBeforePoll)
 }
@@ -871,6 +879,8 @@ func buildConfig(f *mainFlags, des *xtcp_config.EnabledDeserializers) *xtcp_conf
 		S3FlushThresholdJitterPct:    uint32(*f.s3FlushThresholdJitterPct),
 		S3UploadMaxAttempts:          uint32(*f.s3UploadMaxAttempts),
 		S3UploadBackoffCap:           durationpb.New(*f.s3UploadBackoffCap),
+		S3QueueDepth:                 uint32(*f.s3QueueDepth),
+		S3MaxBufferedBytes:           *f.s3MaxBufferedBytes,
 		ReconcileFrequency:           durationpb.New(*f.reconcileFrequency),
 		ReconcileBeforePoll:          *f.reconcileBeforePoll,
 		MaxDumpBytes:                 *f.maxDumpBytes,
@@ -1651,6 +1661,7 @@ func environmentOverrideConfig(c *xtcp_config.XtcpConfig, debugLevel uint) {
 	envOverridePacket(c, debugLevel)
 	envOverrideMarshalAndDest(c, debugLevel)
 	envOverrideCaps(c, debugLevel)
+	envOverrideS3Queue(c, debugLevel)
 	envOverrideKafka(c, debugLevel)
 	envOverrideLabeling(c, debugLevel)
 	envOverrideEnrichment(c, debugLevel)
@@ -2380,6 +2391,19 @@ func checkNoPositional(args []string) error {
 
 // envOverrideCaps applies MAX_DUMP_BYTES, MAX_RECORDS_PER_NETNS and
 // MAX_NETNS; see defineCapFlags.
+// envOverrideS3Queue applies S3_QUEUE_DEPTH and S3_MAX_BUFFERED_BYTES, the
+// s3parquet upload queue's bounds.
+func envOverrideS3Queue(c *xtcp_config.XtcpConfig, debugLevel uint) {
+	if v, ok := envUint32("S3_QUEUE_DEPTH"); ok {
+		c.S3QueueDepth = v
+		logEnv("S3_QUEUE_DEPTH", fmt.Sprintf("c.S3QueueDepth:%d", v), debugLevel)
+	}
+	if v, ok := envUint64("S3_MAX_BUFFERED_BYTES"); ok {
+		c.S3MaxBufferedBytes = v
+		logEnv("S3_MAX_BUFFERED_BYTES", fmt.Sprintf("c.S3MaxBufferedBytes:%d", v), debugLevel)
+	}
+}
+
 func envOverrideCaps(c *xtcp_config.XtcpConfig, debugLevel uint) {
 	if v, ok := envUint64("MAX_DUMP_BYTES"); ok {
 		c.MaxDumpBytes = v
@@ -2436,4 +2460,6 @@ func printConfigCaps(c *xtcp_config.XtcpConfig) {
 	fmt.Println("c.ReconcileTimeout:", c.ReconcileTimeout.AsDuration())
 	fmt.Println("c.MaxNetnsPerOwner:", c.MaxNetnsPerOwner)
 	fmt.Println("c.MaxNetnsAddsPerReconcile:", c.MaxNetnsAddsPerReconcile)
+	fmt.Println("c.S3QueueDepth:", c.S3QueueDepth)
+	fmt.Println("c.S3MaxBufferedBytes:", c.S3MaxBufferedBytes)
 }

cmd/xtcp2/xtcp2_flags_helpers_test.go

@@ -89,7 +89,7 @@ const (
 	// mainFlags field must be a registered flag and every registered flag must
 	// have a field. A flag added without a field, or a field that stops being
 	// registered, is the drift this file exists to notice.
-	mainFlagsFieldCountCst = 139 // + treatmentConfig (embedded agent) and treatmentAttribution (journal reader)
+	mainFlagsFieldCountCst = 141 // + treatmentConfig (embedded agent) and treatmentAttribution (journal reader)
 
 	printFlagsGoldenCst  = "testdata/printflags-golden.txt"
 	defineFlagsGoldenCst = "testdata/defineflags-golden.txt"
@@ -190,6 +190,8 @@ func populatedMainFlags() *mainFlags {
 		s3FlushThresholdJitterPct: u(117),
 		s3UploadMaxAttempts:       u(118),
 		s3UploadBackoffCap:        ms(119),
+		s3QueueDepth:              u(1191),
+		s3MaxBufferedBytes:        u64(1192),
 
 		reconcileFrequency:  ms(120),
 		reconcileBeforePoll: b(false),

cmd/xtcp2/xtcp2_test.go

@@ -316,7 +316,10 @@ func TestEnvOverrideMarshalAndDest(t *testing.T) {
 	t.Setenv("S3_FLUSH_THRESHOLD_JITTER_PCT", "15")
 	t.Setenv("S3_UPLOAD_MAX_ATTEMPTS", "7")
 	t.Setenv("S3_UPLOAD_BACKOFF_CAP", "90s")
+	t.Setenv("S3_QUEUE_DEPTH", "64")
+	t.Setenv("S3_MAX_BUFFERED_BYTES", "1048576")
 	envOverrideMarshalAndDest(c, 0)
+	envOverrideS3Queue(c, 0)
 	if c.MarshalTo != "protoJson" ||
 		c.Dest != "null" || c.DestWriteFiles != 3 {
 		t.Errorf("envOverrideMarshalAndDest mismatch: %+v", c)
@@ -333,6 +336,9 @@ func TestEnvOverrideMarshalAndDest(t *testing.T) {
 	if c.S3UploadBackoffCap.AsDuration() != 90*time.Second {
 		t.Errorf("S3UploadBackoffCap = %v, want 90s", c.S3UploadBackoffCap.AsDuration())
 	}
+	if c.S3QueueDepth != 64 || c.S3MaxBufferedBytes != 1048576 {
+		t.Errorf("S3QueueDepth/S3MaxBufferedBytes = %d/%d, want 64/1048576", c.S3QueueDepth, c.S3MaxBufferedBytes)
+	}
 }
 
 // TestEnvOverrideS3SecretFile covers the Docker `_FILE` convention: the S3
@@ -1048,6 +1054,8 @@ func TestPrintFlags(t *testing.T) {
 	f.s3FlushThresholdJitterPct = &n
 	f.s3UploadMaxAttempts = &n
 	f.s3UploadBackoffCap = &d
+	f.s3QueueDepth = &n
+	f.s3MaxBufferedBytes = &n64
 	f.reconcileFrequency = &d
 	f.reconcileBeforePoll = &b
 	f.maxDumpBytes = &n64
@@ -1278,6 +1286,8 @@ func TestBuildConfig(t *testing.T) {
 		s3FlushThresholdJitterPct: &sftj,
 		s3UploadMaxAttempts:       &suma,
 		s3UploadBackoffCap:        &subc,
+		s3QueueDepth:              &wf,
+		s3MaxBufferedBytes:        &nl,
 		reconcileFrequency:        &rf,
 		reconcileBeforePoll:       &rbp,
 		maxDumpBytes:              &nl,
@@ -1367,6 +1377,8 @@ func TestBuildConfig(t *testing.T) {
 		{"S3FlushThresholdJitterPct", c.S3FlushThresholdJitterPct, uint32(15)},
 		{"S3UploadMaxAttempts", c.S3UploadMaxAttempts, uint32(7)},
 		{"S3UploadBackoffCap", c.S3UploadBackoffCap.AsDuration(), 90 * time.Second},
+		{"S3QueueDepth", c.S3QueueDepth, uint32(11)},
+		{"S3MaxBufferedBytes", c.S3MaxBufferedBytes, uint64(150)},
 		{"ReconcileFrequency", c.ReconcileFrequency.AsDuration(), 3 * time.Minute},
 		{"ReconcileBeforePoll", c.ReconcileBeforePoll, false},
 		{"IpmetaBootstrapPath", c.IpmetaBootstrapPath, ipmetaBootstrap},
@@ -1496,6 +1508,12 @@ func TestConfigValidation_bounds(t *testing.T) {
 		{"boundary s3_upload_max_attempts=100", func(c *xtcp_config.XtcpConfig) { c.S3UploadMaxAttempts = 100 }, false},
 		{"negative s3_upload_max_attempts=0", func(c *xtcp_config.XtcpConfig) { c.S3UploadMaxAttempts = 0 }, true},
 		{"negative s3_upload_max_attempts=101", func(c *xtcp_config.XtcpConfig) { c.S3UploadMaxAttempts = 101 }, true},
+		// s3_queue_depth: lte 4096, 0 = default
+		{"boundary s3_queue_depth=0", func(c *xtcp_config.XtcpConfig) { c.S3QueueDepth = 0 }, false},
+		{"boundary s3_queue_depth=4096", func(c *xtcp_config.XtcpConfig) { c.S3QueueDepth = 4096 }, false},
+		{"negative s3_queue_depth=4097", func(c *xtcp_config.XtcpConfig) { c.S3QueueDepth = 4097 }, true},
+		// s3_max_buffered_bytes: any value, 0 = no byte bound
+		{"positive s3_max_buffered_bytes=64MiB", func(c *xtcp_config.XtcpConfig) { c.S3MaxBufferedBytes = 64 << 20 }, false},
 		// durations: gte 0, with 0 = derive (valid)
 		{"positive s3_flush_interval=45m", func(c *xtcp_config.XtcpConfig) { c.S3FlushInterval = durationpb.New(45 * time.Minute) }, false},
 		{"corner s3_flush_interval=0 (derive)", func(c *xtcp_config.XtcpConfig) { c.S3FlushInterval = durationpb.New(0) }, false},

docs/design-jitter-and-backoff.md

@@ -138,8 +138,8 @@ defer timer.Stop()
 for {
     select {
     case <-d.closedCh:
-        // drain queueCh, finalize, return (unchanged)
-    case item := <-d.queueCh:
+        // drain the queue, finalize, return (unchanged)
+    case <-d.queue.notify:
         processItem(item) // finalizes when accumBytes >= effectiveThreshold (jittered cap)
     case <-timer.C:
         finalize()        // staleness ceiling
@@ -185,7 +185,9 @@ For a `24h` poll: `cap = 24h/10 → clamp → 1h`; 10 attempts of full-jittered
 
 ### Back-pressure & memory
 
-While the worker is in `uploadWithRetry`, it isn't draining `queueCh` (capacity 16). During an outage the queue fills, `Send` blocks, and `flushEnvelope` in the poller blocks — back-pressuring collection. This is *intended* bounded behavior: memory is capped at the queue + one in-flight Parquet object. To keep an outage from stalling collection past one cycle, bound total retry time to roughly one `PollFrequency` — with `base=1s`, `cap=PollFrequency/10`, `attempts=10`, the worst-case sum of windows is ≈ `PollFrequency` (full jitter averages half that). Outages longer than that fall back to today's documented drop-and-continue. Persistent on-disk spooling is explicitly [future work](#open-questions--future-work).
+While the worker is in `uploadWithRetry`, it isn't draining its queue. `Send` never blocks: the queue holds at most `-s3QueueDepth` envelopes (default 16, max 4096) and `-s3MaxBufferedBytes` of them (default 64 MiB, `0` = no byte bound), and past either bound it drops the oldest, counted in `xtcp_s3_dropped_total{reason}` and `xtcp_s3_dropped_bytes_total{reason}`. The newest envelope is always kept. Polling keeps its cadence through an outage; the cost is the oldest data. (Before this bound, a full queue blocked `Send`, and with it `flushEnvelope` mid-poll, stalling collection.)
+
+Memory budget: ≈ `-s3MaxBufferedBytes` (queued envelopes) + the flush threshold (the object being built, ≈ 63 MiB by default) + one object during upload. With the defaults that is about 2–3× the threshold, ≈ 190 MiB; size `GOMEMLIMIT` and the container limit above it. `xtcp_s3_buffered_bytes{stage}` shows the first two. To keep an outage from stalling collection past one cycle, bound total retry time to roughly one `PollFrequency` — with `base=1s`, `cap=PollFrequency/10`, `attempts=10`, the worst-case sum of windows is ≈ `PollFrequency` (full jitter averages half that). Outages longer than that fall back to today's documented drop-and-continue. Persistent on-disk spooling is explicitly [future work](#open-questions--future-work).
 
 ## New configuration surface
 

docs/metrics.md

@@ -200,7 +200,7 @@ All are in `xtcp_counts`.
 | `destS3Parquet` | `uploadRetry` | error | an upload attempt failed and will be retried |
 | `destS3Parquet` | `uploadErrorClass` | `tls`, `timeout`, `canceled`, `dns`, `refused`, `other` | why an attempt failed |
 | `destS3Parquet` | `uploadRowsDropped`, `uploadBytesDropped` | error | rows and bytes lost with a failed upload |
-| `destS3Parquet` | `queueFull` | error | the upload queue was full |
+| `destS3Parquet` | `queueFull` | error | an envelope was dropped from the upload queue, past `-s3QueueDepth` or `-s3MaxBufferedBytes` (`Send` never blocks; see below) |
 | `destS3Parquet` | `unmarshal`, `write`, `writerClose` | error | building the Parquet file failed |
 | `destS3Parquet` | `setS3Flush` | count | the flush settings were changed over gRPC |
 | `destKafka` | `Produce` | count / error | produces and produce failures; also in histograms, as how long a produce took |
@@ -217,6 +217,14 @@ All are in `xtcp_counts`.
 | `destNull`, `destStdout` | `start` | count | sends |
 | marshaller name, such as `protobufListMarshal` | `marshal` | error | the envelope couldn't be marshalled |
 
+The s3parquet upload queue has its own series. Past either bound it drops the oldest envelope, so an S3 outage costs data rather than stalling the poller:
+
+| metric | labels | meaning |
+|---|---|---|
+| `xtcp_s3_dropped_total` | `reason="queue_depth"\|"buffered_bytes"` | envelopes dropped past `-s3QueueDepth` or `-s3MaxBufferedBytes` |
+| `xtcp_s3_dropped_bytes_total` | `reason` | their marshaled bytes |
+| `xtcp_s3_buffered_bytes` | `stage="queue"\|"builder"` | bytes held: envelopes awaiting the worker, and approximate row bytes in the Parquet object being built |
+
 ### Enrichment
 
 `enrich*.go`. Updates per refresh or per event, not per record.