concurrency: single-flight code_map to dedup concurrent cold-cache builds (thundering herd)

Problem

slpl.label() builds a per-codelist label map on first use via _component_label_map(), caching results in the module-level _LABEL_MAP_CACHE. The cache read/write isn't synchronized — the read-then-build-then-write pattern at src/sdmxlib/polars.py:217-242 lets N concurrent threads each see an empty cache, each call cl.code_map(lang=lang) (which runs a DuckDB query through the shared connection), and each write to the cache.

For small codelists this is fine. For CL_CHARACTERISTIC in the StatCan census profile bundle (2,631 codes), one cache build takes ~50-60 seconds against a local DuckLake (the cost is mostly the code_map SQL projection going through the single shared cursor). With N concurrent first-time users they all queue on that cursor and each pays the full cost. Measured:

Concurrent per-thread time
1 ~58 s (first call) / ~0.3 s (cached)
5 ~290 s per thread

Triggered in cenprof-ai's load test (gabrielgellner/cenprof-ai#30). Worked around there by serializing a synthetic query(with_labels=True) at prewarm time, but the upstream fix would benefit anyone with a large codelist + multi-user workload.

Suggested fix

Two clean options:

  1. Per-codelist build lock. Keep one threading.Lock per codelist_urn (or a single global build-lock); the first arrival takes the lock, builds the mapping, populates the cache; concurrent arrivals block until the lock releases, then find the cache hot and return immediately. ~10 lines of code; preserves the current eager-build semantics.

  2. threading.Event per cache key. Slightly more concurrent (different codelists build in parallel) but more bookkeeping. Only worth it if multi-codelist concurrent first-builds become a hot path; probably overkill.

I'd go with (1).

Repro

import threading, time
import sdmxlib.polars as slpl
# … load DSD, open observations frame with CHARACTERISTIC column …

def call():
    t0 = time.monotonic()
    slpl.label(obs.lazy(), dsd, lang="en").collect()
    print(time.monotonic() - t0)

threads = [threading.Thread(target=call) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
# each prints ~290 s

Single-thread comparison: first call ~58 s, subsequent calls ~0.3 s once the cache is populated.