CloudSyncQueue API Reference

Overview

CloudSyncQueue is a persistent store-and-forward queue for edge-to-cloud data synchronization. It is backed by SQLite and designed to survive process crashes, power losses, and network outages. Items remain in the queue until explicitly acknowledged (ack) after successful upload, or negatively acknowledged (nack) for retry with exponential backoff.

The queue supports priority levels (Critical, Anomaly, Telemetry, Logs) and TTL-based expiration. It is thread-safe via an internal threading.RLock.

SQLite Schema

The queue stores items in a single table with an index on priority and retry time:

CREATE TABLE IF NOT EXISTS sync_queue (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    payload TEXT NOT NULL,
    priority INTEGER NOT NULL,
    created_at REAL NOT NULL,
    ttl_seconds INTEGER NOT NULL,
    retry_count INTEGER NOT NULL DEFAULT 0,
    next_retry_at REAL NOT NULL DEFAULT 0
);

CREATE INDEX IF NOT EXISTS idx_sync_queue_priority
    ON sync_queue(priority, next_retry_at);
  

The priority column stores the enum value (1 = Critical, 2 = Anomaly, 3 = Telemetry, 4 = Logs). The next_retry_at column ensures that nacked items are not immediately retried.

Constructor

class CloudSyncQueue:
    def __init__(self, db_path: str = "sync_queue.db") -> None:
        ...
  

On construction, the queue creates the table and index if they do not exist. Each operation opens and closes its own connection; this is slightly less efficient than a persistent connection but safer in multi-process scenarios.

Priority Levels

class Priority(Enum):
    CRITICAL = 1
    ANOMALY = 2
    TELEMETRY = 3
    LOGS = 4
  

Lower numeric values are higher priority. dequeue() and peek() order by priority ASC, next_retry_at ASC, so critical items are always sent first.

Core Methods

enqueue(payload, priority, ttl_seconds)

def enqueue(
    self,
    payload: Dict[str, Any],
    priority: Priority = Priority.TELEMETRY,
    ttl_seconds: int = 86400,
) -> int
  

Returns the auto-incremented row ID of the inserted item.

from pyv_edge_agent.cloud_sync import CloudSyncQueue, Priority

queue = CloudSyncQueue(db_path="/data/sync.db")
item_id = queue.enqueue(
    payload={"sensor": "boiler", "value": 85.0, "alert": "overheat"},
    priority=Priority.CRITICAL,
    ttl_seconds=3600,
)
print(f"Enqueued item {item_id}")
  

dequeue(batch_size)

def dequeue(self, batch_size: int = 10) -> List[QueueItem]
  

Retrieves up to batch_size items that are eligible for transmission: next_retry_at <= now and created_at + ttl_seconds > now. Items are ordered by priority and retry time. This method does not remove items from the queue. Call ack() after successful upload.

items = queue.dequeue(batch_size=50)
for item in items:
    print(f"ID={item.id}, Priority={item.priority.name}, Retry={item.retry_count}")
  

peek(n)

def peek(self, n: int = 5) -> List[QueueItem]
  

Identical to dequeue() in query logic but semantically intended for inspection without transmission intent. Returns up to n eligible items.

up_next = queue.peek(n=3)
for item in up_next:
    print(item.payload)
  

ack(ids)

def ack(self, ids: Sequence[int]) -> int
  

Permanently removes items from the queue. Returns the number of rows deleted. Safe to call with an empty sequence (returns 0).

success_ids = [item.id for item in items]
deleted = queue.ack(success_ids)
print(f"Acknowledged {deleted} items")
  

nack(ids, retry_delay_seconds)

def nack(self, ids: Sequence[int], retry_delay_seconds: float = 60.0) -> int
  

Increments retry_count and sets next_retry_at = now + retry_delay_seconds. Returns the number of rows updated. Use this when upload fails and the item should be retried later.

failed_ids = [item.id for item in items]
updated = queue.nack(failed_ids, retry_delay_seconds=300.0)
print(f"Scheduled {updated} items for retry in 5 minutes")
  

maybe_flush(uploader)

def maybe_flush(self, uploader: Any = None) -> int
  

Flushes all pending items as a single batch. If uploader is provided, it calls uploader.post_batch() with the payloads. On success, items are acked and counters updated. On failure, items are nacked. Returns the number of items flushed.

from pyv_edge_agent.cloud_sync import HTTPCloudUploader

uploader = HTTPCloudUploader(
    endpoint="https://api.pyvorin.com/v1/ingest",
    api_key="pyv_live_xxxxxxxx",
    timeout=30.0,
)
flushed = queue.maybe_flush(uploader=uploader)
print(f"Flushed {flushed} items to cloud")
  

Utility Methods

pending_count()

def pending_count(self) -> int
  

Returns the total number of rows in the sync_queue table, regardless of TTL or retry state.

get_stats()

def get_stats(self) -> Dict[str, Any]
  

Returns queue statistics:

• depth: Total pending count.
• oldest_item_timestamp: Minimum created_at, or None if empty.
• total_retries: Sum of retry_count across all items.
• retrying_items: Count of items with retry_count > 0.

stats = queue.get_stats()
print(f"Depth: {stats['depth']}, Retrying: {stats['retrying_items']}")
  

reset_daily_counters()

def reset_daily_counters(self) -> None
  

Resets messages_sent_today to 0. Should be called by a scheduled job at midnight.

from_dict()

@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "CloudSyncQueue"
  

Factory method for configuration-driven construction:

config = {"db_path": "/data/queue.db"}
queue = CloudSyncQueue.from_dict(config)
  

Locking Behavior

All public methods acquire an internal threading.RLock for the duration of the SQLite transaction. This means:

• Multiple threads can safely enqueue and dequeue concurrently.
• A single thread can recursively acquire the lock (e.g., maybe_flush calling dequeue and ack internally would be safe, though the current implementation keeps all work inside one lock scope).
• The lock does not protect against other processes accessing the same SQLite file. For multi-process deployments, use WAL mode or a separate locking mechanism.

QueueItem Dataclass

@dataclass
class QueueItem:
    id: int
    payload: Dict[str, Any]
    priority: Priority
    created_at: float
    ttl_seconds: int
    retry_count: int = 0
    next_retry_at: float = field(default_factory=time.time)
  

Each QueueItem has a to_dict() method for serialization.

Complete Workflow Example

import time
from pyv_edge_agent.cloud_sync import CloudSyncQueue, Priority, HTTPCloudUploader

queue = CloudSyncQueue(db_path="/tmp/edge_queue.db")

# Enqueue readings from multiple sensors
for i in range(100):
    queue.enqueue(
        payload={"sensor": "motor", "value": float(i), "timestamp": time.time()},
        priority=Priority.TELEMETRY,
    )

# Check queue depth
print(f"Pending: {queue.pending_count()}")

# Peek at next batch
print(queue.peek(n=3))

# Simulate upload
uploader = HTTPCloudUploader(
    endpoint="https://api.pyvorin.com/v1/ingest",
    api_key="pyv_test_xxxxxxxx",
    timeout=10.0,
)
flushed = queue.maybe_flush(uploader=uploader)
print(f"Flushed: {flushed}")

# Final stats
print(queue.get_stats())
  

Queue management

The sections above document the CloudSyncQueue API itself. This section covers the operational side of running the queue in production: delaying first transmission, dead-lettering exhausted items, monitoring queue health, choosing batch sizes, and escalating when uploads keep failing.

Delaying the first transmission attempt

The next_retry_at field is initialised to the current time, meaning a newly enqueued item is immediately eligible for dequeuing. If you want to delay the first transmission attempt — for example, to batch items over a five-minute window — you can set next_retry_at manually by subclassing CloudSyncQueue and overriding enqueue(), or by post-processing the row with a raw SQL UPDATE.

Dead-lettering after max retries

The retry_count field is not currently capped by the queue itself. If you want to dead-letter items after a maximum number of retries, wrap the nack() call in a conditional that checks item.retry_count:

MAX_RETRIES = 5

items = queue.dequeue(batch_size=10)
# ... attempt upload ...
if not success:
    to_retry = []
    to_dead_letter = []
    for item in items:
        if item.retry_count < MAX_RETRIES:
            to_retry.append(item.id)
        else:
            to_dead_letter.append(item)

    if to_retry:
        queue.nack(to_retry, retry_delay_seconds=300.0)

    for item in to_dead_letter:
        write_to_local_log(item.payload)  # preserve for forensic analysis
        queue.ack([item.id])  # remove from queue so it does not block others
  

Monitoring queue health

Operational visibility into the queue is essential for detecting network outages, upstream saturation, or runaway sensor polling. A healthy queue has low depth, a recent oldest item, and few retrying items:

# Simple depth count
depth = queue.pending_count()
print(f"Queue depth: {depth}")

# Full statistics
stats = queue.get_stats()
print(f"Depth: {stats['depth']}")
print(f"Oldest item: {stats['oldest_item_timestamp']}")
print(f"Total retries across all items: {stats['total_retries']}")
print(f"Items currently retrying: {stats['retrying_items']}")
  

If depth grows monotonically over hours, your uploader is failing faster than items are expiring. If retrying_items is high, your network link is flaky or your upstream server is returning 5xx errors.

Batch sizing

The optimal batch size depends on three variables: link bandwidth, upstream latency, and the urgency of the data. There is no universal constant, but the following guidelines have proven effective in production:

• High-bandwidth, low-latency links (fibre, 5G): batch sizes of 500–1000 items maximise throughput and minimise per-request HTTP overhead.
• Low-bandwidth links (2G, LoRaWAN backhaul): batch sizes of 10–50 items prevent timeouts and reduce the blast radius of a single failed upload.
• Critical alarms: bypass batching entirely. Enqueue critical items with Priority.CRITICAL and call dequeue(batch_size=1) in a dedicated fast path that uploads immediately.

Escalation when uploads keep failing

When HTTPCloudUploader.post_batch() returns False, maybe_flush() calls nack() on every item in the batch. The items remain in SQLite with an incremented retry_count and a future next_retry_at. They will not be dequeued again until that time passes, which prevents tight-loop retry storms that could exhaust battery or bandwidth.

If the queue grows large enough to fill the device's storage, you have two escalation options:

1. TTL expiration. Items older than their TTL are silently skipped by dequeue(). They still occupy disk space, but they will never be transmitted. Run a periodic purge job (or extend CloudSyncQueue with a purge_expired() method) to reclaim space.
2. Priority shedding. When disk usage exceeds a threshold, delete all items with priority >= Priority.LOGS while preserving CRITICAL and ANOMALY items.

Tuning parameters

Where to go next

• HTTPCloudUploader Configuration — the component that drains this queue to the cloud
• Offline-First Operation — design patterns built on top of local queuing
• Disaster Recovery — protect queued telemetry against device loss