Collapse the repeated skeleton of the GCS, S3, and Azure row providers #1496

Problem

row_storage_gcs.py, row_storage_s3.py, and row_storage_azure.py each implement the same seven methods (validate_location, probe, put_if_absent, get, stat, delete, and a status or error extractor), each with an async wrapper that hops to a sync twin through asyncio.to_thread, a version-echo check, a create-only enforcement probe, and a timeout budget. Textual duplication is low (pylint reports one block), because the vendor calls differ; the repetition is structural. Two of the sync twins have cyclomatic complexity 19, and a third has 18. A fix to the shared skeleton (a budget rule, a version-echo rule, an error mapping) has to be made three times.

Evidence

All references are to main at 47084e83a, under src/adcp/reporting/ledger/.

• Sizes: row_storage_gcs.py 342 lines, row_storage_s3.py 506, row_storage_azure.py 431.
• Method sets:
  • _GcsProvider (row_storage_gcs.py:114-342): validate_location, probe/_probe, put_if_absent/_put, get/_get, stat/_stat, delete/_delete.
  • _S3Provider (row_storage_s3.py:138-506): the same, plus _assert_bounded_transport, _assert_versioning_enabled, _put_create_only, and _assert_create_only_enforced.
  • _AzureProvider (row_storage_azure.py:151-431): the same, plus _upload and _assert_create_only_enforced.
• Every async method is return await asyncio.to_thread(self._x, ...) (for example row_storage_gcs.py:169, row_storage_s3.py:336, row_storage_azure.py:265).
• Cyclomatic complexity (radon, same commit): _S3Provider._probe 19 (:195), _S3Provider._get 19 (:373), _S3Provider._assert_create_only_enforced 16 (:276), _AzureProvider._get 18 (:293), _AzureProvider._delete 13 (:382), assert_gcs_reporting_row_bucket_policy_v1 14 (row_storage_gcs.py:76).
• row_storage_provider_io.py (111 lines) already exists as the shared helper module for provider I/O.

Suggested fix

Write one base class in row_storage_provider_io.py that owns the to_thread hop, the deadline budget, the version-echo check, and the ReportingRowStoreError mapping, and have each vendor module implement a small set of sync hooks (_do_put, _do_get, _do_stat, _do_delete, _classify_error). The _get methods are the first candidates, since they are the three most complex and differ only in the vendor call and the not-found check.

@bokelley's reply on adcp#7996 notes that obstore is an option behind the interface for the Python side. If that route is taken, obstore's put(..., mode="create") and GetOptions(version=...) replace the three put, get, and delete implementations outright and this issue closes with it.

Introduced in #1468.