feat(master): schedule generic-store WAL merges through the etcd-locked MergeWal task

[beinan]

Problem

Generic stores were never registered with the master. Their MemWAL merges ran only on each worker's own timers (spawn_flush_sweeper / spawn_global_sweeper), so all N workers raced to commit their shard's flushed generations into one base table.

In production (20 workers, ABFS) under storage-account throttling this failed continuously:

Too many concurrent writers. Attempted 2 times, but failed on retry_timeout of 30.000 seconds.
  (lance/src/dataset/write/retry.rs)

Lance's commit-conflict retry re-runs the whole merge_insert and gives up after 30 s wall clock. With 20 writers on one commit point and a throttled object store, the second attempt never completes in time. The generations that did not merge kept the read fan-out — every read opened ~500 generation datasets — which kept the account throttled. Serially triggering /merge-wal by hand while worker timers were disabled still failed on ~30% of shards.

The master's MergeWal task — per-target etcd lock, one task per store, fan-out to WORKER_ENDPOINTS — exists to serialize exactly this, and rollout stores have used it from the start. Generic stores just weren't wired into it.

Fix

Bring generic stores onto the existing master path; no new mechanism.

• MasterState opens _registry.generic.lance (same RolloutRegistry type the data plane writes) alongside the rollout registry; adds generic_uri() / generic_store_options().
• Stats scan enumerates both registries. Generic stores are observed (version, base row count, fragment count, pending WAL generations) via GenericStore::open_existing, and their rows are written under a generic:<name> target. Store names match [A-Za-z0-9_][A-Za-z0-9._-]* so : is unambiguous, and carrying the kind in the target string keeps the etcd task schema, dedupe keys and per-target locks unchanged.
• sweep_merge_wal_inner therefore picks up generic rows over merge_wal_min_generations with no change.
• run_merge_wal parses the target: generic: → POST /api/v1/generic/{name}/merge-wal, bare → /api/v1/internal/merge-wal/{name}.
• Fan-out is now serial instead of join_all. Every worker's merge commits a new version of the same base table, so parallel fan-out was N writers racing one commit point even with the task lock held. This applies to rollout too.
• Compact / IndexId tasks and the compaction sweep remain rollout-only (generic rows are skipped with a clear error if enqueued by hand); retirement only considers rollout rows.

Verification

• cargo clippy --workspace --all-targets -D warnings clean.
• Master suite: 42 default + 25 etcd-backed pass (ETCD_TEST_ENDPOINTS against a local etcd 3.7), including four new tests:
  • merge_wal_routes_generic_targets_to_the_generic_endpoint — a generic:gs task hits /api/v1/generic/gs/merge-wal with the bare name and never the rollout route;
  • merge_wal_calls_workers_serially — three stub workers, shared in-flight counter, asserts max in-flight is 1;
  • sweep_merge_wal_enqueues_over_threshold_and_dedupes — extended: a generic row over threshold is enqueued as generic:ghot alongside the rollout row, and de-dupes on the second sweep;
  • compaction_sweep_skips_generic_rows;
  • scanner generic_store_is_observed_with_pending_wal — 3 sealed adds → row under generic:g with pending_wal_generations == 3, skipped on an unchanged re-scan.

Deployment note: with the master driving generic merges, the worker-side count trigger for generic should be off (ROLLOUT_MERGE_AFTER_GENERATIONS=0) and the 300 s cleanup kept as a fallback — the same posture rollout already runs with.

Refs #256, #257, #258 (supersedes the "unified maintainer" direction there — the right answer was the existing master path), #263.

🤖 Generated with Claude Code