Summary

Replaces the unconditional drop-on-full behavior in the txmeta shard worker pool (introduced in #858) with a mode latched off the Kafka partition high water mark:

• Startup (catching up to the live tail): blocking enqueue with ctx escape so no txmeta cache updates are dropped while the cache is cold. The Kafka consumer is naturally paced by the slowest shard worker.
• Caught up (latched the first time msg.Offset+1 >= msg.HighWaterMark): non-blocking enqueue; a full shard queue is now logged at Warn instead of Error, and the remainder of the batch is abandoned (cache repopulates from Kafka on restart — best-effort by design).

The latch is one-way: once set it stays set, so a long re-catch-up later still operates in live (drop) mode.

Why

In production we see this Error log spamming during scale tests:

ERROR | [kafka_consumer] error processing kafka message on topic txmeta-... skipping: PROCESSING (4): [txmetaHandler] txmeta worker queue full for shard 155

This is the documented best-effort drop firing, so it shouldn't be logged at Error level. And during initial backfill, dropping txmeta updates leaves the cache cold for those entries — better to apply backpressure to Kafka until we're caught up.

Changes

• util/kafka/kafka_consumer.go: added HighWaterMark int64 to KafkaMessage; switched the franz-go fetch loop from EachRecord to EachPartition to read p.HighWatermark; populated HWM in the in-memory consumer via claim.HighWaterMarkOffset(). Field is additive — other consumers ignore it.
• services/subtreevalidation/Server.go: added txmetaCaughtUp atomic.Bool latch.
• services/subtreevalidation/txmetaHandler.go: two-mode enqueueTxmetaWorkItem; maybeMarkTxmetaCaughtUp flips the latch on the partition tail.
• services/subtreevalidation/txmetaHandler_test.go: replaced the now-invalid ReturnsErrorWhenQueueFull test with six tests covering caught-up drop, startup blocking, ctx-cancel unblock, latch flip on HWM, no flip when HWM unset, and one-way latch behavior.

Latch design (deliberate trade-offs)

Documented in code at maybeMarkTxmetaCaughtUp:

• One-way. A node that catches up, runs for hours, then falls badly behind (extended pause) operates in drop mode through the re-catch-up. If you'd rather it re-arm on lag, that's a small follow-up.
• Any-partition. For multi-partition txmeta topics, the latch flips as soon as ANY assigned partition reaches its tail. txmeta is sharded by tx hash and partitions are evenly loaded under normal traffic, so seeing one tail is a strong signal we're broadly caught up. A stricter per-partition gating would extend cold-cache blocking unnecessarily if one partition is temporarily empty or slow.
• Fail-closed on HighWaterMark<=0. An unset HWM (in-memory consumer, hand-built messages) keeps the latch in startup (blocking) mode. Smoke tests are low-throughput, so the shard queues should never fill and the blocking mode is effectively a no-op there.
• Abort-on-first-full-shard preserved. In drop mode the rest of the batch is still abandoned on first full shard. This matches the prior semantic; widening it to "skip full shards but enqueue to others" is a separate change.

Test plan

• go build ./... clean
• go vet ./services/subtreevalidation/... ./util/kafka/... clean
• staticcheck ./services/subtreevalidation/... ./util/kafka/... clean
• golangci-lint run ./services/subtreevalidation/... ./util/kafka/... clean (pre-existing prealloc warnings in unrelated test files)
• go test -race -count=1 ./services/subtreevalidation/... ./util/kafka/... all pass
• Verify in scale-test environment that the production Error log is gone and startup no longer drops txmeta updates while catching up.