Size semantics

README.md

@@ -78,7 +78,36 @@ q.get(task.message_id)
 - Timing metrics. As long as tasks are still in the queue or not pruned, you can see how long they have been there or how long they took to finish.
 - Easy to extend using SQL
 
-## Queue capacity
+## Queue size and capacity
+
+`qsize()`, `empty()`, `full()`, and `maxsize` all use the ready backlog: messages
+that are waiting to be claimed. A message stops contributing to that backlog as
+soon as `pop()` locks it for a worker.
+
+Use the explicit count methods when you need a different view:
+
+- `ready_count()` counts messages waiting to be claimed.
+- `locked_count()` counts messages currently claimed by workers.
+- `done_count()` counts completed messages that have not been pruned.
+- `failed_count()` counts failed messages that have not been pruned.
+- `active_count()` counts ready plus locked messages.
+- `stored_count()` counts every stored row regardless of status.
+
+For example, active locked work does not make the ready backlog non-empty:
+
+```python
+queue = LiteQueue(filename="tasks.sqlite3", maxsize=1)
+queue.put("resize image")
+task = queue.pop()
+
+assert task is not None
+assert queue.qsize() == 0
+assert queue.empty()
+assert not queue.full()
+assert queue.locked_count() == 1
+assert queue.active_count() == 1
+assert queue.stored_count() == 1
+```
 
 `maxsize` limits the number of ready messages and is stored as an immutable
 property of the queue. Omit `maxsize` when reopening a queue to use its stored
@@ -194,6 +223,10 @@ distributions, and uploads them with uv.
 
 ## Important changes
 
+- After version 0.12:
+  - `qsize()` now reports only the ready backlog instead of ready plus locked work.
+  - `empty()`, `full()`, and `maxsize` use the same ready-backlog definition.
+  - Explicit ready, locked, done, failed, active, and stored count methods are available.
 - In version 0.10:
   - Each SQLite database can contain only one LiteQueue queue.
   - You must migrate version 0.9 queues before you use version 0.10 or later.

src/litequeue/__init__.py

@@ -179,7 +179,6 @@ def __init__(
         compiled in serialized mode, as in standard CPython builds. LiteQueue
         serializes writes and explicit transactions while a read-only pool
         serves committed data to other threads.
-
         """
         filename_is_supported = isinstance(filename, (str, Path))
         if not filename_is_supported:
@@ -214,9 +213,7 @@ def __init__(
             cached_statements=0,
             **kwargs,
         )
-
         self.conn.row_factory = sqlite3.Row
-
         self.table = f'"{_QUEUE_TABLE_NAME}"'
 
         with self.transaction(mode="IMMEDIATE"):
@@ -266,7 +263,6 @@ def __init__(
                 f'CREATE UNIQUE INDEX IF NOT EXISTS "Queue_message_id_unique_idx" '
                 f"ON {self.table}(message_id)"
             )
-
             self.conn.execute(
                 f'CREATE INDEX IF NOT EXISTS "Queue_status_message_id_idx" '
                 f"ON {self.table}(status, message_id)"
@@ -282,7 +278,6 @@ def __init__(
                         f"maxsize {validated_maxsize} conflicts with stored maxsize "
                         f"{stored_maxsize} for queue '{_QUEUE_TABLE_NAME}'"
                     )
-
                 effective_maxsize = stored_maxsize
             else:
                 effective_maxsize = validated_maxsize
@@ -382,10 +377,7 @@ def _select_pop_func(self) -> PopFunction:
         return self._pop_transaction
 
     def put(self, data: str) -> Message:
-        """
-        Insert a new message
-        """
-        # timeout: int = None
+        """Insert a new message."""
         message_id = str(uuid7())
         now = time_ns()
 
@@ -498,7 +490,7 @@ def _read_connection(self) -> Iterator[sqlite3.Connection]:
             read_connections.put(read_connection)
 
     def peek(self) -> Message | None:
-        "Show next message to be popped, if any."
+        """Show next message to be popped, if any."""
 
         with self._read_connection() as connection:
             value = connection.execute(
@@ -508,7 +500,7 @@ def peek(self) -> Message | None:
         return _message_from_row(value) if value is not None else None
 
     def get(self, message_id: str) -> Message | None:
-        "Get a message by its `message_id`"
+        """Get a message by its `message_id`."""
 
         with self._read_connection() as connection:
             value = connection.execute(
@@ -576,7 +568,7 @@ def list_locked(self, threshold_seconds: int) -> Iterator[Message]:
                 SELECT * FROM {self.table}
                 WHERE
                   status = {MessageStatus.LOCKED.value}
-                  AND  lock_time < :time_value
+                  AND lock_time < :time_value
                 """.strip(),
                 {"time_value": time_ns() - threshold_nanoseconds},
             ).fetchall()
@@ -585,16 +577,13 @@ def list_locked(self, threshold_seconds: int) -> Iterator[Message]:
             yield _message_from_row(result)
 
     def list_failed(self) -> Iterator[Message]:
-        """
-        Return all the tasks in `FAILED` state.
-        """
+        """Return all the tasks in `FAILED` state."""
 
         with self._read_connection() as connection:
             rows = connection.execute(
                 f"""
                 SELECT * FROM {self.table}
-                WHERE
-                  status = {MessageStatus.FAILED.value}
+                WHERE status = {MessageStatus.FAILED.value}
                 """.strip()
             ).fetchall()
 
@@ -603,7 +592,7 @@ def list_failed(self) -> Iterator[Message]:
 
     def retry(self, message_id: str) -> bool:
         """
-        Mark a locked message as free again.
+        Mark a locked or failed message as ready again.
 
         Return `True` when the message exists, otherwise `False`.
         """
@@ -621,52 +610,58 @@ def retry(self, message_id: str) -> bool:
 
         return cursor.rowcount > 0
 
-    def qsize(self) -> int:
-        """
-        Get current size of the queue.
-        """
+    def _count_statuses(self, *statuses: MessageStatus) -> int:
+        """Return the number of stored messages in the requested states."""
+        if not statuses:
+            return self.stored_count()
 
+        placeholders = ", ".join("?" for _ in statuses)
         with self._read_connection() as connection:
-            cursor = connection.execute(
-                f"""
-            SELECT COUNT(*) FROM {self.table}
-            WHERE status NOT IN ({MessageStatus.DONE.value}, {MessageStatus.FAILED.value})
-            """.strip()
-            )
-            size = next(cursor)[0]
+            value = connection.execute(
+                f"SELECT COUNT(*) FROM {self.table} WHERE status IN ({placeholders})",
+                tuple(status.value for status in statuses),
+            ).fetchone()
+        return int(value[0])
 
-        return size
+    def ready_count(self) -> int:
+        """Return the number of messages waiting to be claimed."""
+        return self._count_statuses(MessageStatus.READY)
 
-    def empty(self) -> bool:
-        """
-        Return True if the queue is empty.
-        """
+    def locked_count(self) -> int:
+        """Return the number of messages currently claimed by workers."""
+        return self._count_statuses(MessageStatus.LOCKED)
 
-        with self._read_connection() as connection:
-            value = connection.execute(
-                f"SELECT COUNT(*) as cnt FROM {self.table} WHERE status = {MessageStatus.READY.value}"
-            ).fetchone()
-        return not bool(value["cnt"])
+    def done_count(self) -> int:
+        """Return the number of completed messages still stored."""
+        return self._count_statuses(MessageStatus.DONE)
 
-    def full(self) -> bool:
-        """
-        Return True if the queue is full.
-        """
+    def failed_count(self) -> int:
+        """Return the number of failed messages still stored."""
+        return self._count_statuses(MessageStatus.FAILED)
 
-        # Here I need to check compared to the maxsize value
-        # If maxsize is not set, the queue can grow forever
-        if self.maxsize is None:
-            return False
+    def active_count(self) -> int:
+        """Return the number of ready and locked messages."""
+        return self._count_statuses(MessageStatus.READY, MessageStatus.LOCKED)
 
+    def stored_count(self) -> int:
+        """Return the total number of rows stored for every message state."""
         with self._read_connection() as connection:
-            value = connection.execute(
-                f"SELECT COUNT(*) as cnt FROM {self.table} WHERE status = {MessageStatus.READY.value}"
-            ).fetchone()
+            value = connection.execute(f"SELECT COUNT(*) FROM {self.table}").fetchone()
+        return int(value[0])
 
-        if value["cnt"] >= self.maxsize:
-            return True
-        else:
+    def qsize(self) -> int:
+        """Return the ready backlog size."""
+        return self.ready_count()
+
+    def empty(self) -> bool:
+        """Return True when the ready backlog is empty."""
+        return self.qsize() == 0
+
+    def full(self) -> bool:
+        """Return True when the ready backlog has reached `maxsize`."""
+        if self.maxsize is None:
             return False
+        return self.qsize() >= self.maxsize
 
     def prune(self, include_failed: bool = True) -> None:
         """

tests/test_size_semantics.py

@@ -0,0 +1,46 @@
+from pathlib import Path
+
+from litequeue import LiteQueue
+
+
+def test_locked_work_is_not_ready_backlog(tmp_path: Path) -> None:
+    queue = LiteQueue(filename=tmp_path / "queue.sqlite3", maxsize=1)
+    queue.put("work")
+
+    locked = queue.pop()
+
+    assert locked is not None
+    assert queue.qsize() == 0
+    assert queue.empty()
+    assert not queue.full()
+    assert queue.ready_count() == 0
+    assert queue.locked_count() == 1
+    assert queue.active_count() == 1
+    assert queue.stored_count() == 1
+
+
+def test_counts_cover_every_message_status(tmp_path: Path) -> None:
+    queue = LiteQueue(filename=tmp_path / "queue.sqlite3")
+    queue.put("locked")
+    queue.put("done")
+    queue.put("failed")
+    queue.put("ready")
+
+    locked = queue.pop()
+    done = queue.pop()
+    failed = queue.pop()
+
+    assert locked is not None
+    assert done is not None
+    assert failed is not None
+    queue.done(done.message_id)
+    queue.mark_failed(failed.message_id)
+
+    assert queue.ready_count() == 1
+    assert queue.locked_count() == 1
+    assert queue.done_count() == 1
+    assert queue.failed_count() == 1
+    assert queue.qsize() == 1
+    assert queue.active_count() == 2
+    assert queue.stored_count() == 4
+    assert not queue.empty()