|
| 1 | +"""Optional process-local write fencing, not a backend handoff acknowledgement. |
| 2 | +
|
| 3 | +Only an explicitly attached base.control_write_gate enables this foundation. |
| 4 | +No configuration flag, receipt publisher or controller release is provided yet. |
| 5 | +""" |
| 6 | + |
| 7 | +import asyncio |
| 8 | +import functools |
| 9 | +import inspect |
| 10 | +import math |
| 11 | +import threading |
| 12 | +import time |
| 13 | + |
| 14 | + |
| 15 | +class ControlWriteBlocked(RuntimeError): |
| 16 | + """A closed, expired or wrong-provider epoch cannot admit another mutation.""" |
| 17 | + |
| 18 | + |
| 19 | +class ControlWriteGate: |
| 20 | + """Fence transport attempts across component threads and independent event loops. |
| 21 | +
|
| 22 | + Closing admission is permanent. Uncertain remote outcomes remain latched, |
| 23 | + even after all local tasks finish. A transport drain is never proof that a |
| 24 | + Hub is physically fenced or that a backend transition was committed. |
| 25 | + """ |
| 26 | + |
| 27 | + def __init__(self, provider, epoch, config_version, expires_at, clock=time.monotonic): |
| 28 | + """Bind one gate to a controller epoch and a monotonic lease deadline.""" |
| 29 | + if not provider or not epoch or isinstance(config_version, bool) or not isinstance(config_version, int): |
| 30 | + raise ValueError("Invalid controller binding") |
| 31 | + if not math.isfinite(expires_at) or expires_at <= clock(): |
| 32 | + raise ValueError("Invalid controller lease") |
| 33 | + self.provider = provider |
| 34 | + self.epoch = epoch |
| 35 | + self.config_version = config_version |
| 36 | + self.expires_at = expires_at |
| 37 | + self._clock = clock |
| 38 | + self._condition = threading.Condition() |
| 39 | + self._closed = False |
| 40 | + self._transition_id = None |
| 41 | + self._next_token = 0 |
| 42 | + self._pending = set() |
| 43 | + self._uncertain = set() |
| 44 | + |
| 45 | + def admit(self, provider): |
| 46 | + """Atomically admit one transport attempt or reject before transmission.""" |
| 47 | + with self._condition: |
| 48 | + if self._closed or provider != self.provider or self._clock() >= self.expires_at: |
| 49 | + raise ControlWriteBlocked("Controller epoch cannot admit a mutation") |
| 50 | + self._next_token += 1 |
| 51 | + self._pending.add(self._next_token) |
| 52 | + return self._next_token |
| 53 | + |
| 54 | + def complete(self, token, definitive): |
| 55 | + """Finish local work, retaining every remotely ambiguous outcome.""" |
| 56 | + with self._condition: |
| 57 | + if token not in self._pending: |
| 58 | + raise ValueError("Unknown or already completed mutation") |
| 59 | + self._pending.remove(token) |
| 60 | + if definitive is not True: |
| 61 | + self._uncertain.add(token) |
| 62 | + self._condition.notify_all() |
| 63 | + |
| 64 | + def close(self, epoch, config_version, transition_id): |
| 65 | + """Permanently close this exact epoch for one transition, idempotently.""" |
| 66 | + with self._condition: |
| 67 | + if epoch != self.epoch or config_version != self.config_version or not transition_id: |
| 68 | + raise ValueError("Wrong controller binding") |
| 69 | + if self._transition_id is not None and transition_id != self._transition_id: |
| 70 | + raise ValueError("Controller already belongs to another transition") |
| 71 | + self._closed = True |
| 72 | + self._transition_id = transition_id |
| 73 | + self._condition.notify_all() |
| 74 | + |
| 75 | + def snapshot(self): |
| 76 | + """Return local evidence without claiming a durable or physical handoff.""" |
| 77 | + with self._condition: |
| 78 | + return { |
| 79 | + "provider": self.provider, |
| 80 | + "epoch": self.epoch, |
| 81 | + "config_version": self.config_version, |
| 82 | + "transition_id": self._transition_id, |
| 83 | + "closed": self._closed, |
| 84 | + "expired": self._clock() >= self.expires_at, |
| 85 | + "pending": len(self._pending), |
| 86 | + "uncertain": len(self._uncertain), |
| 87 | + } |
| 88 | + |
| 89 | + def wait_drained(self, epoch, config_version, transition_id, timeout): |
| 90 | + """Wait for local drain only; timeout, expiry and ambiguity return false. |
| 91 | +
|
| 92 | + This synchronous method belongs on a worker thread, never on a component |
| 93 | + event loop needed to finish admitted work. Cancelling an async caller |
| 94 | + must not cause it to emit an acknowledgement from a finally block. |
| 95 | + """ |
| 96 | + if not math.isfinite(timeout) or timeout < 0: |
| 97 | + raise ValueError("Invalid drain timeout") |
| 98 | + deadline = time.monotonic() + timeout |
| 99 | + with self._condition: |
| 100 | + if not self._closed or self._transition_id != transition_id or epoch != self.epoch or config_version != self.config_version: |
| 101 | + raise ValueError("Drain requires the exact closed transition") |
| 102 | + while self._pending: |
| 103 | + remaining = min(deadline - time.monotonic(), self.expires_at - self._clock()) |
| 104 | + if remaining <= 0: |
| 105 | + return False |
| 106 | + self._condition.wait(remaining) |
| 107 | + return not self._uncertain and self._clock() < self.expires_at |
| 108 | + |
| 109 | + |
| 110 | +def control_gate_for(base): |
| 111 | + """Use only an explicitly attached gate; ordinary installations are unchanged.""" |
| 112 | + gate = getattr(base, "__dict__", {}).get("control_write_gate") |
| 113 | + if gate is not None and not isinstance(gate, ControlWriteGate): |
| 114 | + raise ControlWriteBlocked("Invalid controller gate") |
| 115 | + return gate |
| 116 | + |
| 117 | + |
| 118 | +def control_mutation(provider, endpoint, definitive_result, definitive_error): |
| 119 | + """Fence each actual Cloud mutation, including each retry's transport call.""" |
| 120 | + |
| 121 | + def decorate(function): |
| 122 | + """Wrap a transport without changing its public calling convention.""" |
| 123 | + signature = inspect.signature(function) |
| 124 | + |
| 125 | + @functools.wraps(function) |
| 126 | + async def guarded(self, *args, **kwargs): |
| 127 | + """Track one submitted write until its complete classified outcome.""" |
| 128 | + gate = control_gate_for(getattr(self, "base", None)) |
| 129 | + if gate is None: |
| 130 | + return await function(self, *args, **kwargs) |
| 131 | + arguments = signature.bind(self, *args, **kwargs) |
| 132 | + arguments.apply_defaults() |
| 133 | + if arguments.arguments.get("endpoint") != endpoint or not arguments.arguments.get("post"): |
| 134 | + return await function(self, *args, **kwargs) |
| 135 | + token = gate.admit(provider) |
| 136 | + definitive = False |
| 137 | + try: |
| 138 | + result = await function(self, *args, **kwargs) |
| 139 | + definitive = definitive_result(result) is True |
| 140 | + return result |
| 141 | + except Exception as error: |
| 142 | + definitive = definitive_error(error) is True |
| 143 | + raise |
| 144 | + finally: |
| 145 | + # BaseException (including cancellation) deliberately stays uncertain. |
| 146 | + gate.complete(token, definitive) |
| 147 | + |
| 148 | + return guarded |
| 149 | + |
| 150 | + return decorate |
| 151 | + |
| 152 | + |
| 153 | +async def gated_mqtt_publish(gate, provider, publish, owner_loop): |
| 154 | + """Keep owner-loop work counted even if its awaiting caller is cancelled. |
| 155 | +
|
| 156 | + Broker acceptance is not physical completion. Every submitted MQTT mutation |
| 157 | + stays uncertain until a future protocol can prove the physical fence. |
| 158 | + """ |
| 159 | + token = gate.admit(provider) |
| 160 | + |
| 161 | + def finished(future): |
| 162 | + """Consume completion and retain an unresolved physical outcome.""" |
| 163 | + if not future.cancelled(): |
| 164 | + future.exception() |
| 165 | + gate.complete(token, False) |
| 166 | + |
| 167 | + coroutine = None |
| 168 | + try: |
| 169 | + coroutine = publish() |
| 170 | + if owner_loop is not None and owner_loop is not asyncio.get_running_loop(): |
| 171 | + future = asyncio.run_coroutine_threadsafe(coroutine, owner_loop) |
| 172 | + future.add_done_callback(finished) |
| 173 | + waiter = asyncio.wrap_future(future) |
| 174 | + else: |
| 175 | + future = asyncio.ensure_future(coroutine) |
| 176 | + future.add_done_callback(finished) |
| 177 | + waiter = future |
| 178 | + except BaseException: |
| 179 | + if coroutine is not None: |
| 180 | + coroutine.close() |
| 181 | + gate.complete(token, False) |
| 182 | + raise |
| 183 | + await asyncio.shield(waiter) |
0 commit comments