DX-818: Add realtime transcription support with failover handling#450
DX-818: Add realtime transcription support with failover handling#450yadavsahil197 wants to merge 4 commits into
Conversation
Broly Security ScanNote Summary 6 actionable finding(s) in this PR
5 highest-priority actionable rows in the table below (critical/high first, then top medium).
Dismiss false positivesEach dismissable row has a Dismiss key (
Note Re-scan this PR anytime with
|
blainekasten
left a comment
blainekasten
left a comment
There was a problem hiding this comment.
This looks great!
As we discussed in slack, let's move this functionality under the beta namespace for now to give us some wiggle room if we decide to change the api interface in the near term.
85d5879 to
6d913ab
Compare
|
Moved to the beta namespace as discussed! |
yadavsahil197
changed the title
| for _held, reclaim in holders: | ||
| if deficit <= 0: | ||
| return | ||
| deficit -= reclaim(deficit) |
There was a problem hiding this comment.
I think there's a cross-thread mutation hazard here. charge() runs on whatever thread is appending, and when the pool is under pressure this ends up calling reclaim → AudioBuffer.trim_to on other sessions buffers. But those sessions are running on their own event loops so we can popleft from a deque while its owner's writer loop is mid-iteration in read_from which will be a error.
It would only fire under pool pressure with multiple sessions - could we marshal the reclaim onto the holder's loop via call_soon_threadsafe or put a lock around AudioBuffer?
| pool=pool, | ||
| ) | ||
| self._pool = pool | ||
| pool.register(self, lambda: self.state.buffer.size, self.state.reclaim) |
There was a problem hiding this comment.
Registering in init but only unregistering in close()/_fail() means a session that's constructed but never started (or just abandoned) stays strongly referenced by the client-scoped pool forever. Could we move registration into start(), or make _holders a WeakKeyDictionary so the pool never keeps a session alive on its own?
| # 300s no-append idle timeout | ||
| if self._keepalive_silence and self._last_append_at: | ||
| if now - self._last_append_at > 200.0: | ||
| await self.append(b"\x00" * int(self.state.bps * 0.1)) # 100ms of silence |
There was a problem hiding this comment.
This append() can raise a RealtimeBufferOverflowError when overflow="error" or a stored terminal failure racing in via _raise_if_failed() and _watchdog_loop only handles CancelledError. So the watchdog task dies with an unobserved exception and we lose echo probing + keepalive while the session looks healthy. Worth wrapping the loop body in an except Exception that at least logs.
| self._writer_wakeup.set() | ||
| for task in (self._watchdog_task, self._writer_task, self._reconnect_task, self._reader_task): | ||
| if task is not None: | ||
| task.cancel() |
There was a problem hiding this comment.
We cancel these but never await them, and in the sync the loop gets stopped right after so tasks can be destroyed while still pending (asyncio warnings, and the reader may not finish tearing down the socket). A gather(*tasks, return_exceptions=True) after the cancels would make shutdown deterministic.
| self._session = self._call(_create()) | ||
|
|
||
| def __enter__(self) -> "RealtimeTranscriptionSession": | ||
| self.start() |
There was a problem hiding this comment.
Probably want a try/except around this that calls close() before re-raising. If start() raises here (e.g. bad base_url → RealtimeConnectionError), exit never runs so the background loop thread we spawned in init lives until atexit.
3 participants
No description provided.