lace

163@dataclass(frozen=True)
164class BoundedInterlaceOutcome:
165    terminal: Terminal
166    observations: list[InterlaceEvent]
167    peer_by: str | None
168    outgoing_encoding: str | None
169    incoming_encoding: str | None

186class Cancellation:
187    """Thread-safe caller cancellation for bounded operations.
188
189    ``set()`` may be called from any thread. The operation retires its machine
190    and transport, settles started store work through position publication,
191    then raises ``concurrent.futures.CancelledError``. ``set()`` wakes blocked
192    waits directly; nothing polls it.
193    """
194
195    def __init__(self) -> None:
196        self._lock = threading.Lock()
197        self._set = False
198        self._wakes = _WakeHub()
199
200    def set(self) -> None:
201        with self._lock:
202            if self._set:
203                return
204            self._set = True
205        self._wakes.signal()
206
207    def is_set(self) -> bool:
208        with self._lock:
209            return self._set
210
211    def _subscribe(
212        self, condition: threading.Condition | None = None
213    ) -> _WakeSubscription:
214        """A wake already signalled when cancellation was set before subscribing."""
215        subscription = self._wakes.subscribe(condition)
216        if self.is_set():
217            subscription.signal(raise_errors=False)
218        return subscription

46@dataclass(frozen=True)
47class Closed(InterlaceEvent):
48    reason: CloseReason

15@dataclass(frozen=True)
16class CompactionReport:
17    """What one completed compaction removed and released.
18
19    ``boundary`` is the high-water the retention sweep evaluated.
20    ``bytes_released`` counts arena bytes between the new and previous
21    committed end; ``release`` is ``"unsupported"`` when the filesystem cannot
22    release them, after the root is already consistent.
23    """
24
25    boundary: int
26    retained: int
27    removed: int
28    bytes_released: int
29    release: Literal["ok", "unsupported"]

115@dataclass(frozen=True)
116class Complete:
117    kind: Literal["complete"] = "complete"

158class ConnectedPeer:
159    """One host-owned connected participant over established ILTP byte IO."""
160
161    def __init__(self) -> None:
162        raise TypeError("ConnectedPeer instances are created by transport factories")
163
164    @classmethod
165    def from_duplex(
166        cls,
167        io: RawIo,
168        *,
169        role: CarrierRole,
170        by_secret: str | None = None,
171        peer_proof: PeerProofAssurance | ProofAssurance = PeerProofAssurance.DISABLED,
172        channel_binding: bytes | bytearray | memoryview | None = None,
173        expected_peer_by: str | None = None,
174        stream_encodings: Sequence[str] | None = None,
175        port: str | None = None,
176    ) -> "ConnectedPeer":
177        """Adopt one duplex POSIX IO owner as a connected peer."""
178        return cls._from_owners(
179            io,
180            io,
181            role=role,
182            by_secret=by_secret,
183            peer_proof=peer_proof,
184            channel_binding=channel_binding,
185            expected_peer_by=expected_peer_by,
186            stream_encodings=stream_encodings,
187            port=port,
188            duplex=True,
189        )
190
191    @classmethod
192    def from_split(
193        cls,
194        reader: RawIo,
195        writer: RawIo,
196        *,
197        role: CarrierRole,
198        by_secret: str | None = None,
199        peer_proof: PeerProofAssurance | ProofAssurance = PeerProofAssurance.DISABLED,
200        channel_binding: bytes | bytearray | memoryview | None = None,
201        expected_peer_by: str | None = None,
202        stream_encodings: Sequence[str] | None = None,
203        port: str | None = None,
204    ) -> "ConnectedPeer":
205        """Adopt independent POSIX reader/writer owners as one connected peer."""
206        return cls._from_owners(
207            reader,
208            writer,
209            role=role,
210            by_secret=by_secret,
211            peer_proof=peer_proof,
212            channel_binding=channel_binding,
213            expected_peer_by=expected_peer_by,
214            stream_encodings=stream_encodings,
215            port=port,
216            duplex=False,
217        )
218
219    @classmethod
220    def _from_owners(
221        cls,
222        reader: RawIo,
223        writer: RawIo,
224        *,
225        role: CarrierRole,
226        by_secret: str | None,
227        peer_proof: PeerProofAssurance | ProofAssurance,
228        channel_binding: bytes | bytearray | memoryview | None,
229        expected_peer_by: str | None,
230        stream_encodings: Sequence[str] | None,
231        port: str | None,
232        duplex: bool,
233    ) -> "ConnectedPeer":
234        from ._raw_split_transport import _RawTransport
235
236        try:
237            assurance, binding, encodings = cls._validate_inputs(
238                role, by_secret, peer_proof, channel_binding, expected_peer_by,
239                stream_encodings, port,
240            )
241            transport = _RawTransport(reader, writer, role=role, duplex=duplex)
242        except BaseException as error:
243            owners = (reader,) if reader is writer else (writer, reader)
244            _raise_after_cleanup(error, *(lambda owner=owner: owner.close() for owner in owners))
245        # The complete transport now owns cleanup, including failed peer setup.
246        return cls._from_transport(
247            transport, by_secret=by_secret, peer_proof=assurance,
248            channel_binding=binding, expected_peer_by=expected_peer_by,
249            stream_encodings=encodings, port=port,
250        )
251
252    @classmethod
253    def _from_transport(
254        cls,
255        transport: _ConnectedTransport,
256        *,
257        by_secret: str | None,
258        peer_proof: PeerProofAssurance,
259        channel_binding: bytes | None = None,
260        expected_peer_by: str | None = None,
261        stream_encodings: list[str] | None = None,
262        port: str | None = None,
263    ) -> "ConnectedPeer":
264        """Adopt a transport with already validated and snapshotted options."""
265        try:
266            peer = cls.__new__(cls)
267            peer._initialize(
268                transport, by_secret=by_secret, peer_proof=peer_proof,
269                channel_binding=channel_binding, expected_peer_by=expected_peer_by,
270                stream_encodings=stream_encodings, port=port,
271            )
272        except BaseException as error:
273            _raise_after_cleanup(error, transport.close)
274        return peer
275
276    @staticmethod
277    def _validate_inputs(
278        role: CarrierRole,
279        by_secret: str | None,
280        peer_proof: PeerProofAssurance | ProofAssurance,
281        channel_binding: bytes | bytearray | memoryview | None,
282        expected_peer_by: str | None,
283        stream_encodings: Sequence[str] | None,
284        port: str | None,
285    ) -> tuple[PeerProofAssurance, bytes | None, list[str] | None]:
286        if role not in {"listener", "connector"}:
287            raise ValueError("transport carrier_role must be listener or connector")
288        if by_secret is not None:
289            _native.Record.by_from_secret(by_secret)
290        if not isinstance(peer_proof, str):
291            raise TypeError("peer_proof must be a PeerProofAssurance or string")
292        assurance = PeerProofAssurance(peer_proof)
293        binding = None
294        if assurance is PeerProofAssurance.CHANNEL_BOUND:
295            if by_secret is None:
296                raise ValueError("channel-bound proof requires by_secret")
297            if channel_binding is None:
298                raise ValueError("channel-bound proof requires channel_binding")
299            if not isinstance(channel_binding, (bytes, bytearray, memoryview)):
300                raise TypeError("channel_binding must be bytes, bytearray, or memoryview")
301            binding = bytes(channel_binding)
302            if len(binding) != 32:
303                raise ValueError("channel_binding must contain exactly 32 bytes")
304        elif channel_binding is not None:
305            raise ValueError("channel_binding requires channel-bound proof")
306        if expected_peer_by is not None:
307            _native.validate_by(expected_peer_by)
308            if assurance is PeerProofAssurance.DISABLED:
309                raise ValueError("expected_peer_by requires enabled peer_proof")
310        encodings = None
311        if stream_encodings is not None:
312            if isinstance(stream_encodings, (str, bytes)) or not isinstance(
313                stream_encodings, Sequence
314            ):
315                raise TypeError("stream_encodings must be a sequence of token strings")
316            encodings = list(stream_encodings)
317            if not all(isinstance(token, str) for token in encodings):
318                raise TypeError("stream_encodings must be a sequence of token strings")
319            _native.validate_stream_encodings(encodings)
320        if port is not None:
321            if not isinstance(port, str):
322                raise TypeError("port must be a string")
323            _native.validate_port(port)
324        return assurance, binding, encodings
325
326    def _initialize(
327        self,
328        transport: _ConnectedTransport,
329        *,
330        by_secret: str | None,
331        peer_proof: PeerProofAssurance,
332        channel_binding: bytes | None,
333        expected_peer_by: str | None,
334        stream_encodings: list[str] | None,
335        port: str | None,
336    ) -> None:
337        nonce_seed = os.urandom(32)
338        if not any(nonce_seed):
339            raise LaceIntegrationError(
340                "OS randomness returned an all-zero carrier nonce seed"
341            )
342        self._input: _ConnectedInput | None = _ConnectedInput(
343            transport, transport.carrier_role, by_secret,
344            peer_proof, channel_binding, nonce_seed, expected_peer_by,
345            stream_encodings, port,
346        )
347        self._close_error: BaseException | None = None
348
349    def _claim(self) -> _ConnectedInput:
350        if self._input is None:
351            raise RuntimeError("connected peer is closed or already owned by an interlace")
352        owned, self._input = self._input, None
353        return owned
354
355    def close(self) -> None:
356        if self._input is not None:
357            owned, self._input = self._input, None
358            try:
359                owned.close()
360            except BaseException as error:
361                self._close_error = error
362        if self._close_error is not None:
363            raise self._close_error
364
365    def __enter__(self) -> "ConnectedPeer":
366        return self
367
368    def __exit__(
369        self,
370        exc_type: type[BaseException] | None,
371        exc: BaseException | None,
372        tb: TracebackType | None,
373    ) -> None:
374        try:
375            self.close()
376        except BaseException as cleanup_error:
377            failure = _combine_failures(exc, cleanup_error)
378            assert failure is not None
379            raise failure

125@dataclass(frozen=True)
126class Failed:
127    error: InterlaceError
128    kind: Literal["failed"] = "failed"

41@dataclass(frozen=True)
42class FixedPoint(InterlaceEvent):
43    pass

52@dataclass(frozen=True)
53class FieldExclusion:
54    """Records carrying a matching ``Field`` occurrence; ``None`` matches anything."""
55    name: str | None = None
56    index: str | None = None
57    value: str | None = None

528class ForEachStream:
529    """Observer stream: records saved to the private observer, plus boundaries."""
530
531    def __init__(self, opened: OpenInterlace) -> None:
532        self._opened = opened
533
534    def close(self) -> None:
535        self._opened.close()
536
537    def next_event(self, *, timeout: float | None = None) -> InterlaceEvent | None:
538        timeout = _validate_timeout(timeout)
539        deadline = None if timeout is None else time.monotonic() + timeout
540        while True:
541            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
542            event = self._opened.next_event(timeout=remaining)
543            if (
544                event is None
545                or isinstance(event, (FixedPoint, Closed))
546                or (isinstance(event, RecordSaved) and event.position == "first")
547            ):
548                return event
549
550    def __enter__(self) -> "ForEachStream":
551        return self
552
553    def __exit__(
554        self,
555        exc_type: type[BaseException] | None,
556        exc: BaseException | None,
557        tb: TracebackType | None,
558    ) -> None:
559        self._opened.__exit__(exc_type, exc, tb)
560
561    def __iter__(self) -> Iterator[InterlaceEvent]:
562        return self
563
564    def __next__(self) -> InterlaceEvent:
565        observation = self.next_event()
566        if observation is None:
567            raise StopIteration
568        return observation

156@dataclass(frozen=True)
157class GetOutcome:
158    terminal: Terminal
159    observations: list[InterlaceEvent]
160    record: Record | None

36class InterlaceError(LaceError):
37    category: ClassVar[ErrorCategory] = "integration"
38
39    def __init__(
40        self,
41        message: str,
42        *,
43        position: ParticipantPosition | None = None,
44        record_hash: str | None = None,
45        phase: TimeoutPhase | None = None,
46    ) -> None:
47        super().__init__(message)
48        self.message = message
49        self.position = position
50        self.record_hash = record_hash
51        self.phase = phase

18@dataclass(frozen=True)
19class InterlaceEvent:
20    pass

245class Lace:
246    """Non-clonable Python owner of one adopted store and its publication state."""
247
248    def __init__(
249        self,
250        store: _StoreOwner,
251        *,
252        position_may_advance_externally: bool,
253        operand_cache: int = DEFAULT_OPERAND_CACHE_ENTRIES,
254        _adoption_token: object | None = None,
255    ):
256        if _adoption_token is not _ADOPTION_TOKEN:
257            raise TypeError(
258                "construct Lace with a built-in store constructor, or use lace.advanced.with_store"
259            )
260        if type(position_may_advance_externally) is not bool:
261            raise TypeError("position_may_advance_externally must be a bool")
262        self._store = store
263        self._position_may_advance_externally = position_may_advance_externally
264        try:
265            if isinstance(operand_cache, bool) or not isinstance(operand_cache, int):
266                raise TypeError("operand_cache must be an int")
267            if operand_cache < 0:
268                raise ValueError("operand_cache must not be negative")
269            # Peer operands held by hash, shared by this Lace's connected operations.
270            self._operand_cache = _native.OperandCache(operand_cache)
271            try:
272                initial_position = store.current_position()
273            except Exception as error:
274                if isinstance(error, LaceError):
275                    raise
276                raise LaceStoreError(str(error)) from error
277            self._closed = False
278            self._close_complete = False
279            self._close_error: BaseException | None = None
280            # Registration order is close order.
281            self._open_interlaces: dict[OpenInterlace, None] = {}
282            self._mutation_lock = threading.Lock()
283            self._position_lock = threading.Lock()
284            self._latest_store_position = initial_position
285            self._position_subscriptions: set[_StorePositionSubscription] = set()
286            self._work_condition = threading.Condition()
287            self._wakes = _WakeHub()
288            self._active_store_work: set[PendingWork] = set()
289            self._unacknowledged_work_errors: list[BaseException] = []
290            self._terminal_failure: BaseException | None = None
291            self._next_poll_at = time.monotonic() + _REMOTE_POSITION_POLL_INTERVAL_SECONDS
292            self._poll_in_flight = False
293        except BaseException as error:
294            _raise_after_cleanup(error, self._store_close)
295
296    @classmethod
297    def _adopt(
298        cls,
299        store: _StoreOwner,
300        *,
301        position_may_advance_externally: bool,
302        operand_cache: int = DEFAULT_OPERAND_CACHE_ENTRIES,
303    ) -> "Lace":
304        return cls(
305            store,
306            position_may_advance_externally=position_may_advance_externally,
307            operand_cache=operand_cache,
308            _adoption_token=_ADOPTION_TOKEN,
309        )
310
311    @classmethod
312    def memory(
313        cls,
314        records: Record | Iterable[Record] | None = None,
315        *,
316        operand_cache: int = DEFAULT_OPERAND_CACHE_ENTRIES,
317    ) -> "Lace":
318        return cls._with_initial_records(
319            cls._adopt(
320                _indexed_memory_owner(),
321                position_may_advance_externally=False,
322                operand_cache=operand_cache,
323            ),
324            records,
325        )
326
327    if hasattr(_native.NativeStore, "filesystem"):
328        @classmethod
329        def filesystem(
330            cls,
331            path: str | os.PathLike[str],
332            records: Record | Iterable[Record] | None = None,
333            *,
334            operand_cache: int = DEFAULT_OPERAND_CACHE_ENTRIES,
335        ) -> "Lace":
336            return cls._with_initial_records(
337                cls._adopt(
338                    _blocking_native_owner(_native.NativeStore.filesystem, path),
339                    position_may_advance_externally=False,
340                    operand_cache=operand_cache,
341                ),
342                records,
343            )
344
345    if hasattr(_native.NativeStore, "lmdb"):
346        @classmethod
347        def lmdb(
348            cls,
349            path: str | os.PathLike[str],
350            *,
351            operand_cache: int = DEFAULT_OPERAND_CACHE_ENTRIES,
352        ) -> "Lace":
353            return cls._adopt(
354                _blocking_native_owner(_native.NativeStore.lmdb, path),
355                position_may_advance_externally=True,
356                operand_cache=operand_cache,
357            )
358
359    @staticmethod
360    def _with_initial_records(
361        lace: "Lace", records: Record | Iterable[Record] | None
362    ) -> "Lace":
363        if records is None:
364            return lace
365        try:
366            from .operations import Failed, StoppedByLimit
367            from ._porcelain import store
368
369            outcome = store(lace, records)
370            if isinstance(outcome.terminal, Failed):
371                raise outcome.terminal.error
372            if isinstance(outcome.terminal, StoppedByLimit):
373                raise LaceLimitExceededError("initial record admission stopped by a limit")
374            return lace
375        except BaseException as error:
376            _raise_after_cleanup(error, lace.close)
377
378    def handle(self) -> LaceHandle:
379        self._raise_if_unavailable()
380        return LaceHandle(self)
381
382    def __reduce_ex__(self, protocol: SupportsIndex) -> NoReturn:
383        # copy, deepcopy and pickle all reduce through this.
384        raise TypeError("Lace is non-clonable; clone Lace.handle() instead")
385
386    def _register_open_interlace(self, interlace: OpenInterlace) -> None:
387        with self._work_condition:
388            self._raise_if_unavailable_locked()
389            self._open_interlaces[interlace] = None
390
391    def _unregister_open_interlace(self, interlace: OpenInterlace) -> None:
392        with self._work_condition:
393            self._open_interlaces.pop(interlace, None)
394
395    def _subscribe_wakes(
396        self, condition: threading.Condition | None = None
397    ) -> _WakeSubscription:
398        with self._work_condition:
399            self._raise_if_unavailable_locked()
400            return self._wakes.subscribe(condition)
401
402    def _subscribe_store_position(
403        self, condition: threading.Condition
404    ) -> _StorePositionSubscription:
405        subscription = _StorePositionSubscription(self, condition)
406        self._wakes.attach(subscription)
407        with self._work_condition:
408            try:
409                self._raise_if_unavailable_locked()
410            except BaseException:
411                subscription.close()
412                raise
413            with self._position_lock:
414                self._position_subscriptions.add(subscription)
415        return subscription
416
417    def _unsubscribe_store_position(
418        self, subscription: _StorePositionSubscription
419    ) -> None:
420        with self._position_lock:
421            self._position_subscriptions.discard(subscription)
422
423    def _raise_if_unavailable_locked(self) -> None:
424        if self._terminal_failure is not None:
425            raise self._terminal_failure
426        if self._closed:
427            raise RuntimeError("Lace is closed")
428
429    def _raise_if_unavailable(self) -> None:
430        with self._work_condition:
431            self._raise_if_unavailable_locked()
432
433    def _poison(self, error: BaseException) -> BaseException:
434        with self._work_condition:
435            if self._terminal_failure is None:
436                for execution in self._active_store_work:
437                    execution.prevent_start(error)
438                self._terminal_failure = error
439            self._poll_in_flight = False
440            retained = self._terminal_failure
441            self._work_condition.notify_all()
442        self._wakes.signal()
443        return retained
444
445    def _latest_position(self) -> int:
446        with self._work_condition:
447            if self._terminal_failure is not None:
448                raise self._terminal_failure
449        with self._position_lock:
450            return self._latest_store_position
451
452    def _publish_store_position(self, position: int) -> None:
453        """Retain and signal one confirmed position while the mutation gate is held."""
454        with self._work_condition:
455            if self._terminal_failure is not None:
456                raise self._terminal_failure
457            with self._position_lock:
458                if position < self._latest_store_position:
459                    failure = LaceIntegrationError(
460                        f"store position regressed from {self._latest_store_position} to {position}"
461                    )
462                    subscriptions: tuple[_StorePositionSubscription, ...] = ()
463                elif position == self._latest_store_position:
464                    return
465                else:
466                    failure = None
467                    self._latest_store_position = position
468                    subscriptions = tuple(self._position_subscriptions)
469
470        signal_error: BaseException | None = None
471        for subscription in subscriptions:
472            try:
473                subscription.signal()
474            except BaseException as error:
475                signal_error = _combine_failures(signal_error, error)
476        if signal_error is not None:
477            failure = LaceIntegrationError(
478                f"failed to publish confirmed store position: {signal_error}"
479            )
480            failure.__cause__ = signal_error
481        if failure is None:
482            return
483        retained = self._poison(failure)
484        raise retained
485
486    def _begin_store_work(self, execution: PendingWork) -> None:
487        with self._work_condition:
488            self._raise_if_unavailable_locked()
489            self._active_store_work.add(execution)
490
491    def _finish_store_work(self, execution: PendingWork) -> None:
492        with self._work_condition:
493            self._active_store_work.discard(execution)
494            self._work_condition.notify_all()
495
496    def _retain_store_work_error(self, error: BaseException) -> None:
497        with self._work_condition:
498            self._unacknowledged_work_errors.append(error)
499            self._work_condition.notify_all()
500
501    def _external_position_poll_deadline(self, now: float) -> float | None:
502        with self._work_condition:
503            self._raise_if_unavailable_locked()
504            if not self._position_may_advance_externally:
505                return None
506            if self._poll_in_flight and self._next_poll_at <= now:
507                return now + _REMOTE_POSITION_POLL_INTERVAL_SECONDS
508            return self._next_poll_at
509
510    def _start_due_external_position_refresh(self, now: float) -> bool:
511        from ._store_execution import PendingWork
512
513        with self._work_condition:
514            self._raise_if_unavailable_locked()
515            if (
516                not self._position_may_advance_externally
517                or self._poll_in_flight
518                or now < self._next_poll_at
519            ):
520                return False
521            execution = PendingWork.refresh(self)
522            self._poll_in_flight = True
523            self._next_poll_at = now + _REMOTE_POSITION_POLL_INTERVAL_SECONDS
524            self._active_store_work.add(execution)
525        try:
526            execution.start_thread()
527        except BaseException as error:
528            failure = LaceIntegrationError(
529                f"failed to start external store-position probe: {error}"
530            )
531            failure.__cause__ = error
532            with self._work_condition:
533                # An interrupted start may already have handed the probe to
534                # its worker. Keep that owner until publication/settlement.
535                if execution.channel.claim():
536                    self._active_store_work.discard(execution)
537                self._poll_in_flight = False
538                self._next_poll_at = (
539                    time.monotonic() + _REMOTE_POSITION_POLL_INTERVAL_SECONDS
540                )
541                if self._terminal_failure is None:
542                    for pending in self._active_store_work:
543                        pending.prevent_start(failure)
544                    self._terminal_failure = failure
545                retained = self._terminal_failure
546                self._work_condition.notify_all()
547            self._wakes.signal()
548            if not isinstance(error, Exception):
549                raise
550            raise retained
551        return True
552
553    def _finish_external_position_refresh(
554        self,
555        execution: PendingWork,
556        failure: BaseException | None,
557    ) -> None:
558        with self._work_condition:
559            self._active_store_work.discard(execution)
560            self._poll_in_flight = False
561            self._next_poll_at = (
562                time.monotonic() + _REMOTE_POSITION_POLL_INTERVAL_SECONDS
563            )
564            if failure is not None and self._terminal_failure is None:
565                for pending in self._active_store_work:
566                    pending.prevent_start(failure)
567                self._terminal_failure = failure
568            self._work_condition.notify_all()
569        self._wakes.signal()
570
571    def _settle_external_position_poll(self) -> None:
572        """Finish the one Lace-owned probe before an operation releases this Lace."""
573        subscription = self._wakes.subscribe()
574        failure: BaseException | None = self._terminal_failure
575        try:
576            while True:
577                try:
578                    # Publish retired predecessors before waiting on a queued
579                    # probe; never consume another live operation's response.
580                    self._drain_store_work(wait=False)
581                    self._finish_ready_position_refreshes()
582                    with self._work_condition:
583                        refresh_pending = any(
584                            execution.is_refresh for execution in self._active_store_work
585                        )
586                        failure = _combine_failures(failure, self._terminal_failure)
587                    if not refresh_pending:
588                        break
589                    subscription.wait()
590                except BaseException as error:
591                    failure = _combine_failures(failure, error)
592        finally:
593            try:
594                subscription.close()
595            except BaseException as error:
596                failure = _combine_failures(failure, error)
597        if failure is not None:
598            raise failure
599
600    def _has_ready_store_work(self) -> bool:
601        with self._work_condition:
602            return any(
603                (execution.is_refresh or execution.retired) and execution.ready()
604                for execution in self._active_store_work
605            )
606
607    def _finish_ready_position_refreshes(self) -> bool:
608        """Publish ready Lace-local refreshes without taking another operation's work."""
609        with self._work_condition:
610            refreshes = tuple(
611                execution
612                for execution in self._active_store_work
613                if execution.is_refresh
614            )
615        progressed = False
616        for refresh in refreshes:
617            if refresh.finished or not refresh.ready():
618                continue
619            refresh.finish(self, wait=False)
620            progressed = refresh.finished or progressed
621        return progressed
622
623    def _drain_store_work(
624        self,
625        *,
626        wait: bool,
627        excluding: set[PendingWork] | None = None,
628        report_errors: bool = False,
629    ) -> None:
630        excluded = excluding or set()
631        while True:
632            with self._work_condition:
633                executions = tuple(
634                    execution
635                    for execution in self._active_store_work
636                    if execution not in excluded
637                )
638            if not executions:
639                break
640            progressed = False
641            for execution in executions:
642                if not wait and (not execution.retired or not execution.ready()):
643                    continue
644                execution.finish(self, wait=wait)
645                progressed = True
646                execution.retain_late_failure(self)
647            if not wait:
648                break
649            if not progressed:
650                raise RuntimeError("active StoreWork could not be drained")
651        if report_errors and self._unacknowledged_work_errors:
652            failure: BaseException | None = None
653            for error in self._unacknowledged_work_errors:
654                failure = _combine_failures(failure, error)
655            self._unacknowledged_work_errors.clear()
656            if failure is not None:
657                raise failure
658
659    def _wait_for_store_work(self) -> None:
660        subscription = self._wakes.subscribe()
661        failure: BaseException | None = None
662        try:
663            while True:
664                try:
665                    with self._work_condition:
666                        executions = tuple(self._active_store_work)
667                    if not executions:
668                        break
669                    progressed = False
670                    for execution in executions:
671                        if not execution.ready():
672                            continue
673                        try:
674                            error = execution.finish(self, wait=False)
675                            if execution.executed:
676                                failure = _combine_failures(failure, error)
677                        finally:
678                            if execution.finished:
679                                execution.retain_late_failure(self)
680                        progressed = True
681                    if not progressed:
682                        subscription.wait()
683                except BaseException as error:
684                    failure = _combine_failures(failure, error)
685            for retained_error in self._unacknowledged_work_errors:
686                failure = _combine_failures(failure, retained_error)
687            self._unacknowledged_work_errors.clear()
688        finally:
689            try:
690                subscription.close()
691            except BaseException as error:
692                failure = _combine_failures(failure, error)
693        if failure is not None:
694            raise failure
695
696    def _store_close(self) -> None:
697        try:
698            self._store.close()
699        except LaceError:
700            raise
701        except BaseException as error:
702            raise LaceStoreError(str(error)) from error
703
704    def close(self) -> None:
705        # Every claimed operation is released here, whatever interrupts close.
706        claimed: list[OpenInterlace] = []
707        try:
708            self._close_claimed(claimed)
709        finally:
710            for interlace in claimed:
711                interlace._release()
712
713    def _close_claimed(self, claimed: list[OpenInterlace]) -> None:
714        with self._work_condition:
715            if self._closed:
716                while not self._close_complete:
717                    try:
718                        self._work_condition.wait()
719                    except BaseException as error:
720                        self._close_error = _combine_failures(self._close_error, error)
721                if self._close_error is not None:
722                    raise self._close_error
723                return
724            # Claim every live operation before the participant lifecycle
725            # changes; one that another caller is using makes close fail busy.
726            for interlace in list(self._open_interlaces):
727                interlace._claim()
728                claimed.append(interlace)
729            # Stop queued calls before the close cut; already-executing calls
730            # still own their mandatory refresh and publication interval.
731            for execution in self._active_store_work:
732                execution.prevent_start(RuntimeError("Lace is closed"))
733            # This is the close linearization point. New work and operation
734            # registration fail, while counted work retains the store.
735            self._closed = True
736        failure: BaseException | None = self._terminal_failure
737        try:
738            for interlace in claimed:
739                try:
740                    interlace._finish()
741                except BaseException as error:
742                    failure = _combine_failures(failure, error)
743            with self._work_condition:
744                self._open_interlaces.clear()
745            try:
746                self._wait_for_store_work()
747            except BaseException as error:
748                failure = _combine_failures(failure, error)
749            failure = _combine_failures(failure, self._terminal_failure)
750            try:
751                self._store_close()
752            except BaseException as error:
753                failure = _combine_failures(failure, error)
754            failure = _combine_failures(failure, self._terminal_failure)
755        finally:
756            with self._work_condition:
757                self._close_error = _combine_failures(failure, self._close_error)
758                self._close_complete = True
759                self._work_condition.notify_all()
760        if self._close_error is not None:
761            raise self._close_error
762
763    def __enter__(self) -> "Lace":
764        return self
765
766    def __exit__(
767        self,
768        exc_type: type[BaseException] | None,
769        exc: BaseException | None,
770        tb: TracebackType | None,
771    ) -> None:
772        try:
773            self.close()
774        except BaseException as cleanup_error:
775            if exc is None:
776                raise
777            _combine_failures(exc, cleanup_error)
778            raise exc