9898#: worth stalling the app's exit.
9999_SHUTDOWN_CLOSE_TIMEOUT = 2.0
100100
101+ #: How long the recreate loop waits for the transport before looking
102+ #: again. The transport tells us when it reconnects, so this is only
103+ #: the self-heal path for a notification that never arrives; it must
104+ #: never be the *only* way we wake up, and nothing may depend on it
105+ #: being short.
106+ _TRANSPORT_RECHECK_SECONDS = 30.0
107+
108+ #: A channel that died sooner than this means something is wrong
109+ #: (an unreachable node) rather than a drive having finished.
110+ _CHANNEL_SHORT_LIFE_SECONDS = 2.0
111+
112+ #: Backoff bounds for re-offering after a channel dies immediately.
113+ _RECREATE_BACKOFF_MIN_SECONDS = 2.0
114+ _RECREATE_BACKOFF_MAX_SECONDS = 60.0
115+
101116
102117def _make_key () -> str :
103118 """Mint (or read) this run's automation key."""
@@ -125,6 +140,12 @@ def __init__(self) -> None:
125140 #: Set at app shutdown so the recreate loop stops offering
126141 #: fresh channels while the runtime is going away.
127142 self ._shutting_down = False
143+ #: Wakes the recreate loop when the transport reconnects.
144+ #: A wakeup *only*: what gates the loop is asking the
145+ #: transport where it is, so a set() we miss costs a
146+ #: re-check delay rather than a device that never becomes
147+ #: drivable again.
148+ self ._transport_connected = asyncio .Event ()
128149 #: Supplied by the caller, which has the private-api access
129150 #: to read it (baplus may not reach ``_babase``).
130151 self ._app_instance_id = ''
@@ -161,15 +182,23 @@ def on_transport_connected(
161182
162183 self ._app_instance_id = app_instance_id
163184 ws_url = _ws_url_for (node_base_url )
185+
186+ # Before the early-out below, not after: a task that parked
187+ # because the transport was down is still very much running,
188+ # and this is the wakeup that un-parks it.
189+ self ._transport_connected .set ()
190+
164191 if ws_url == self ._node_url and self ._task is not None :
165- # Same node, still running; nothing to do.
192+ # Same node, still running; nothing to do. (Also the
193+ # ordinary path back from a transport outage: the task
194+ # notices the node it holds is reachable again.)
166195 return
167196
168197 self ._stop ()
169198 self ._node_url = ws_url
170199 if self ._key is None :
171200 self ._key = _make_key ()
172- self ._task = asyncio .create_task (self ._run (ws_url ))
201+ self ._task = asyncio .create_task (self ._run ())
173202
174203 def _stop (self ) -> None :
175204 """Tear down any live channel."""
@@ -213,7 +242,7 @@ async def shutdown(self) -> None:
213242 except BaseException : # pylint: disable=broad-except
214243 pass # Cancelled (or already failing) -- we're leaving.
215244
216- async def _run (self , ws_url : str ) -> None :
245+ async def _run (self ) -> None :
217246 """Offer a channel, and a fresh one each time one ends.
218247
219248 A SmartSocket channel is one device + one driver over its
@@ -224,8 +253,30 @@ async def _run(self, ws_url: str) -> None:
224253 new one under a new id + locator. A single driver's own wifi
225254 blip is invisible to this loop -- the endpoint resumes the
226255 same channel internally and only returns here on a real end.
256+
257+ Offering is gated on the transport being connected. Not the
258+ live channel -- an endpoint mid-resume keeps its full
259+ reconnect budget, since a brief drop is exactly what it is
260+ built to ride out and killing it would cost a driver its
261+ session. What stops is the *recreating*: with the device
262+ asleep or the network gone, its node is unreachable by
263+ definition, and re-offering into that produces a fresh
264+ locator and a fresh round of failures every couple of
265+ seconds, forever, for nobody.
227266 """
267+ backoff = _RECREATE_BACKOFF_MIN_SECONDS
228268 while not self ._shutting_down :
269+ ws_url = await self ._await_transport ()
270+ if ws_url is None :
271+ return # Shutting down.
272+ # Adopt whatever node the transport is on now. Normally
273+ # ``on_transport_connected`` restarts us on a node change
274+ # and this is simply the url we already had; taking it
275+ # from the transport each time means a notification we
276+ # somehow miss costs a recheck interval instead of
277+ # leaving us dialing a node nobody is on any more.
278+ self ._node_url = ws_url
279+
229280 started = time .monotonic ()
230281 channel_dead = await self ._run_one_channel (ws_url )
231282 if not channel_dead or self ._shutting_down :
@@ -237,10 +288,51 @@ async def _run(self, ws_url: str) -> None:
237288 # racing to read the fresh locator right after ending the
238289 # old one must not find a gap. Back off only when a
239290 # channel dies almost immediately, which means something
240- # is wrong (an unreachable node) rather than a drive
241- # having finished.
242- if time .monotonic () - started < 2.0 :
243- await asyncio .sleep (2.0 )
291+ # is wrong (a node that answers but won't hold a channel)
292+ # rather than a drive having finished.
293+ if time .monotonic () - started < _CHANNEL_SHORT_LIFE_SECONDS :
294+ await asyncio .sleep (backoff )
295+ backoff = min (backoff * 2.0 , _RECREATE_BACKOFF_MAX_SECONDS )
296+ else :
297+ backoff = _RECREATE_BACKOFF_MIN_SECONDS
298+
299+ async def _await_transport (self ) -> str | None :
300+ """Wait until we have a node to offer a channel on.
301+
302+ Returns its attach url, or ``None`` if we're shutting down.
303+
304+ Level-triggered on purpose: we ask the transport where it is
305+ rather than trusting a remembered flag, and the wait always
306+ times out. Both halves are about recovery -- a gate that can
307+ only be re-opened by an event arriving is a gate that strands
308+ the device for the rest of the run if one ever doesn't, and
309+ that failure would look exactly like automation being broken.
310+ """
311+ while not self ._shutting_down :
312+ # Clear *before* looking, so a connect landing between
313+ # the two leaves the event set and the wait returns at
314+ # once. Clearing after would be the classic missed-wakeup
315+ # race, costing a full recheck interval.
316+ self ._transport_connected .clear ()
317+ ws_url = self ._connected_node_ws_url ()
318+ if ws_url is not None :
319+ return ws_url
320+ try :
321+ await asyncio .wait_for (
322+ self ._transport_connected .wait (),
323+ timeout = _TRANSPORT_RECHECK_SECONDS ,
324+ )
325+ except TimeoutError :
326+ pass
327+ return None
328+
329+ def _connected_node_ws_url (self ) -> str | None :
330+ """Our transport's current node as an attach url, if any."""
331+ plus = babase .app .plus
332+ if plus is None :
333+ return None
334+ base_url = plus .cloud .get_connected_node_base_url ()
335+ return None if base_url is None else _ws_url_for (base_url )
244336
245337 async def _run_one_channel (self , ws_url : str ) -> bool :
246338 """Hold one channel until it dies. True if it died on its own.
0 commit comments