Fix: enable WAL and batch writes in TelemetryStore to avoid SQLITE_BUSY under concurrency (fixes #560)#597
Open
SoulSniper-V2 wants to merge 3 commits into
Open
Conversation
SoulSniper-V2
requested review from
ANarayan,
jonsaadfalcon and
robbym-dev
as code owners
There was a problem hiding this comment.
Pull request overview
This PR updates TelemetryStore’s SQLite usage to better tolerate concurrent write load by enabling WAL mode and batching inserts, reducing commit/fsync frequency and aiming to prevent SQLITE_BUSY errors under inference concurrency.
Changes:
- Enable SQLite WAL mode and set
synchronous=NORMALonTelemetryStoreinitialization. - Buffer telemetry + mining stats rows in-memory and flush with
executemany()when the batch threshold is reached (and on query/close). - Add a unit test to verify that commits are deferred until the batch size is hit.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 6 comments.
| File | Description |
|---|---|
src/openjarvis/telemetry/store.py |
Enables WAL + adds lock-protected batching/flush logic for telemetry/mining inserts. |
tests/telemetry/test_store.py |
Adds a regression test asserting batched writes delay persistence until flush/threshold. |
Comments suppressed due to low confidence (1)
src/openjarvis/telemetry/store.py:315
- _select_dicts() executes SQL without holding the store lock. To keep the single sqlite3.Connection safe under multi-threaded access (even with check_same_thread=False), reads should also be serialized with the same lock used for writes/flush.
def _select_dicts(self, sql: str, params: tuple[Any, ...]) -> list[dict[str, Any]]:
cur = self._conn.execute(sql, params)
columns = [desc[0] for desc in cur.description]
return [dict(zip(columns, row)) for row in cur.fetchall()]
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+149
to
+153
| def __init__(self, db_path: str | Path, batch_size: int = 50) -> None: | ||
| self._db_path = str(db_path) | ||
| self._conn = sqlite3.connect(self._db_path, check_same_thread=False) | ||
| self._conn.execute("PRAGMA journal_mode=WAL") | ||
| self._conn.execute("PRAGMA synchronous=NORMAL") |
Comment on lines
+159
to
+162
| self._lock = threading.Lock() | ||
| self._batch_size = batch_size | ||
| self._telemetry_batch: list[tuple[Any, ...]] = [] | ||
| self._mining_batch: list[tuple[Any, ...]] = [] |
Comment on lines
301
to
304
| def close(self) -> None: | ||
| """Close the underlying SQLite connection.""" | ||
| self.flush() | ||
| self._conn.close() |
Comment on lines
308
to
310
| def _fetchall(self, sql: str = "SELECT * FROM telemetry") -> list: | ||
| self.flush() | ||
| return self._conn.execute(sql).fetchall() |
Comment on lines
+160
to
+170
| import sqlite3 | ||
|
|
||
| conn = sqlite3.connect(db_path) | ||
| assert len(conn.execute("SELECT * FROM telemetry").fetchall()) == 0 | ||
| conn.close() | ||
|
|
||
| # Hit batch size | ||
| store.record(rec) | ||
| conn = sqlite3.connect(db_path) | ||
| assert len(conn.execute("SELECT * FROM telemetry").fetchall()) == 2 | ||
| conn.close() |
Comment on lines
+152
to
+155
| def test_batching_delays_commit(self, tmp_path: Path) -> None: | ||
| db_path = tmp_path / "test.db" | ||
| store = TelemetryStore(db_path, batch_size=2) | ||
| rec = TelemetryRecord(timestamp=time.time(), model_id="m1", engine="e1") |
Collaborator
|
Thank you @SoulSniper-V2! Can you fix the tests and merge conflicts? |
Contributor
Author
|
Hey @ElliotSlusky! Fixed — merge conflicts are resolved and all 13 tests pass (including the new WAL and concurrency tests from main). I also integrated main's |
…ches Serialize all SQLite access under the store lock, flush pending batches before queries, and add a stale-batch flush interval so external readers like TelemetryAggregator see committed rows. Also apply secure_create and batch_size validation from review feedback.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What does this PR do?
Resolves Issue #560 and fully closes out parent Issue #219 alongside PR #570.
The
TelemetryStorewas previously missingPRAGMA journal_mode=WALwhich made it susceptible toSQLITE_BUSYerrors under concurrent inference load. Furthermore, every telemetry insert was immediately callingcommit(), forcing high fsync I/O overhead.Changes
PRAGMA journal_mode=WALandPRAGMA synchronous=NORMALwhen initializingTelemetryStoreso it correctly mimicsTraceStore.threading.Lock()toTelemetryStoreto buffer incoming records inself._telemetry_batchandself._mining_batch.record()andrecord_mining_stats()to hold onto records until thebatch_size(default: 50) is reached.flush()method that uses SQLite's fastexecutemany()operation. This method is called automatically when hitting the batch limit, when querying (list_recent), and when closing the DB connection.test_batching_delays_committotest_store.pyto ensure batching logic correctly defers disk commits.Testing
Fixes Issue
Fixes #560