feat(spanner): KPEP-0001 backend with full storage.Interface conformance + TTL

[zachsmith1]

Summary

Stands up the in-tree Spanner backend for KPEP-0001 (pluggable storage backends) and brings it to a production-ready state: 24/24 upstream storage.Interface conformance, ordered watch delivery, and a three-layer TTL eviction design.

Depends on kplane-dev/kubernetes#4 (factory.Backend registry seam) and #5 (DecodeCallback at storage package), both merged.

What lands

Phase 1 — KPEP-0001 backend wiring (earlier commits on this branch)

• Storage backend registry (registry/registry.go) with a Backend interface and BuildFactoryBackend() hook
• Spanner backend moved in-tree (backends/spanner/) with the aggregator pattern
• BackendFactory hook on DecoratorConfig so the multicluster decorator dispatches through it
• factory.Backend adapter (backends/spanner/factory_backend.go) so a single registration covers every factory.Create callsite in the apiserver — CR storage, master/peer endpoint leases, service IP/NodePort allocators
• Options.Build() auto-applies the schema on first use, so a fresh database doesn't require a separate schema-apply step

Phase 2 — Ticket+heap broadcaster (broadcast.go)

Replaces the channel-fanout broadcaster with a per-write Ticket abstraction plus a single-goroutine heap dispatcher.

Why: Spanner commit timestamps come from TrueTime and are assigned in commit order, but concurrent Apply() calls return out-of-order at the client. Publishing in arrival order produced non-monotonic event RVs to watchers, violating the cacher watchCache's binary-search invariant.

How: AcquireWrite() reserves a Ticket BEFORE Apply; defer Cancel covers every error path; Publish on success enqueues the event in a min-heap. The dispatcher waits until pendingTickets==0 before flushing, at which point TrueTime guarantees no future Apply can produce an RV below anything buffered.

Verified by broadcast_order_test.go (concurrent-write monotonicity) and cacher_list_test.go (cacher integration).

Phase 3 — storage.Interface conformance

Brings the backend to the upstream RunTest* contract from staging/.../pkg/storage/testing. Hooked up in conformance_test.go.

Substantive fixes:

• storageKeyFromSpannerKey: preserve the leading slash. Was stripping /registry/ AND the leading /, so the cacher's ListPrefix lookup missed every item it indexed — the load-bearing bug behind "empty LIST returns".
• Get + GetList: guard too-high RVs with TooLargeResourceVersionError instead of letting Spanner return DeadlineExceeded.
• Watch: honor ctx.Done() in both the pre-loop and streaming-loop paths.
• Watch (legacy RV=0): send initial events as ADDED to match etcd's areInitialEventsRequired behavior; skip the trailing bookmark (which belongs to the explicit WatchList contract); advance startRev to max(initialItemRVs) (advancing to bookmarkRV races concurrent writes that committed just before GetCurrentResourceVersion and silently filters them out).
• GetList: ensure non-nil empty slice when predicate doesn't match.
• resolveListRV(): one helper for the three callsites that need a guaranteed-nonzero list RV — returns an explicit error instead of silently substituting 0.

Existing watch tests updated to pass the post-create RV; the previous expectation (no initial events on a watch with no RV specified) only held because of the bug being fixed.

Phase 4 — TTL eviction (config.go, store.go, broadcast.go)

Three independent layers, each doing one thing well:

Layer	Mechanism
Read correctness	expire_at IS NULL OR expire_at > CURRENT_TIMESTAMP() predicate on every Get/List SELECT — reads never return expired rows
Watch correctness	Per-broadcaster TTL scanner publishes synthetic Delete events for newly-expired rows; CAS-verifies under a write Ticket before publishing (closes the refresh race)
Physical cleanup	Spanner's ROW DELETION POLICY (OLDER_THAN(expire_at, INTERVAL 0 DAY)) — Google-managed background deletion, zero client cost

Per-replica scanner; no cross-replica coordination needed because broadcasters are disjoint and watchers are local. From the watcher's POV, a synthetic Delete is identical to a real Delete — same path through processEvent, same prevValue decoding, same watch.Deleted emission.

Test plan

• go build ./... clean
• go test ./backends/spanner/ -run "^TestConformance_" — 24/24 PASS (upstream RunTest* battery)
• go test ./backends/spanner/ -run "^Test[^C]" — all existing tests PASS
• End-to-end apiserver bootstrap against Spanner matches the etcd baseline: 55 rows, LIST behavior identical, PostStartHooks complete
• CI green

Why one PR

The four phases overlap heavily in broadcast.go / store.go / watcher.go — splitting them would create PRs that individually break tests until the others land. Commits within the PR are organized by phase for review.