Fixув SQlite duplicate telemetry race condition

[lucas-souza-enerlab]

Fix SQLite storage duplicate telemetry race condition

Summary

When using SQLite event storage ("type": "sqlite"), the gateway sends duplicate telemetry to ThingsBoard. Every reading appears twice with timestamps 1–10 ms apart. Switching to "type": "memory" eliminates the issue.

Root cause

The Database background thread pre-fetches the next batch of records into an in-memory cache (__next_batch). The pre-fetch is unlocked by can_prepare_new_batch(), which is called inside get_event_pack() — before event_pack_processing_done() deletes the records from the database.

This creates a race window:

Thread A (gateway)                    Thread B (Database)
─────────────────                     ───────────────────
get_event_pack()
  → SELECT id, msg FROM ... LIMIT 10
  → returns batch1 [msg_0..msg_9]
  → can_prepare_new_batch()  ───────→ flag set, pre-fetch triggered
                                      → SELECT ... LIMIT 10
                                      → caches [msg_0..msg_9] (same records!)
  ... send via MQTT ...
event_pack_processing_done()
  → DELETE WHERE id <= 10
get_event_pack()
  → read_data() returns stale cache
  → returns [msg_0..msg_9] again  ←── DUPLICATE

Memory storage is not affected because Queue.get_nowait() is a destructive read — once consumed, data cannot be re-read.

Fix

Move can_prepare_new_batch() from get_event_pack() to event_pack_processing_done(), after delete_data(). This ensures the Database thread only pre-fetches the next batch after the current batch has been deleted.

Before (buggy):

# sqlite_event_storage.py

def get_event_pack(self):
    data_from_storage = self.read_data()
    # ... process data ...
    self.__read_database.can_prepare_new_batch()  # ← unlocks pre-fetch TOO EARLY
    return event_pack_messages

def event_pack_processing_done(self):
    if not self.stopped.is_set():
        self.delete_data(self.delete_time_point)  # ← records deleted AFTER pre-fetch

After (fixed):

# sqlite_event_storage.py

def get_event_pack(self):
    data_from_storage = self.read_data()
    # ... process data ...
    return event_pack_messages  # ← can_prepare_new_batch() removed

def event_pack_processing_done(self):
    if not self.stopped.is_set():
        self.delete_data(self.delete_time_point)
        self.__read_database.can_prepare_new_batch()  # ← moved HERE, after deletion

Commits

#	Commit	Description
1	2b057d8b	Regression test that fails on the original code, proving the bug exists
2	ddb4ae76	The fix — test now passes

Reviewers can checkout commit 1 and run the test to reproduce the bug independently:

git checkout 2b057d8b
python -m pytest tests/unit/service/test_sqlite_duplicate_race_condition.py -v -s
# → SQLite test FAILS (duplicates detected), memory test PASSES

Then checkout commit 2 to verify the fix:

git checkout ddb4ae76
python -m pytest tests/unit/service/test_sqlite_duplicate_race_condition.py -v -s
# → Both tests PASS

Test plan

• Regression test reproduces the race condition on original code (commit 1 fails)
• Fix resolves the race condition (commit 2 passes)
• Memory storage control test passes on both commits
• Existing test_sqlite_storage in tests/unit/service/test_storage.py still passes
• Manual validation with a real connector over extended runtime

Environment

• ThingsBoard IoT Gateway version: 3.8.2
• Python: 3.13
• OS: Debian (Docker)

[CLAassistant]

All committers have signed the CLA.

[harkatos]

Good fix, very needed to use SQLite

[lucasamoreira28]

Thanks, I was facing the same problem and I tested your solution. It solved my issue.

[samson0v]

@lucas-souza-enerlab, many thanks for your contribution!