Recover the publisher lease after ClickHouse errors and wait for a crashed predecessor at start

native/csrc/catalog/bindings_store.cpp

@@ -79,6 +79,8 @@ dc::StorageServiceConfig service_config(const py::dict& d) {
   c.writer.publish_timeout_ns =
       get<uint64_t>(d, "publish_timeout_ns", c.writer.publish_timeout_ns);
   c.writer.clock_skew_ns = get<uint64_t>(d, "clock_skew_ns", c.writer.clock_skew_ns);
+  c.start_lease_wait_ns =
+      get<uint64_t>(d, "start_lease_wait_ns", c.start_lease_wait_ns);
   c.indexer.max_packs = get<int>(d, "indexer_max_packs", c.indexer.max_packs);
   c.indexer.max_estimated_bytes =
       get<uint64_t>(d, "indexer_max_estimated_bytes", c.indexer.max_estimated_bytes);
@@ -112,6 +114,12 @@ py::dict snapshot_dict(const dc::StorageServiceSnapshot& s) {
   out["swept_on_start"] = s.swept_on_start;
   out["pending_index"] = s.pending_index;
   out["rejected_packs"] = s.rejected_packs;
+  out["failed"] = s.failed;
+  out["lease_state"] = s.lease_state;
+  // Seconds on the monotonic clock, comparable with time.monotonic() (both
+  // CLOCK_MONOTONIC on Linux); 0.0 when not quarantined.
+  out["quarantined_until"] = static_cast<double>(s.quarantined_until_ns) / 1e9;
+  out["lease_reacquisitions"] = s.lease_reacquisitions;
   out["last_error"] = s.last_error;
   return out;
 }