queue

queue

The outbox: commands waiting to reach a device, and what came back.

Three questions this answers that the protocol alone cannot (spec §44/§45):

• what happens when the target is not there — and the honest answer differs per tool. Opening a screen on a sleeping phone is pointless by the time it wakes (REQUIRE_ONLINE); a notification is worth keeping (QUEUE_UNTIL_EXPIRATION); some things are simply not worth retrying (DROP_IF_OFFLINE);
• whether a command already ran — replaying a delivery must never produce two effects, so results are recorded against the idempotency key and replayed rather than re-executed;
• what the user is told — every terminal state carries a French sentence that is true. A queued command says queued. It never says done.

The one rule above all (spec §57): an offline device is never reported as having executed anything.

Classes

CommandQueue

CommandQueue(db_path: str | Path | None = None)

Durable record of every command this device sent or received.

Source code in src/diapason/mesh/queue.py

def __init__(self, db_path: str | Path | None = None) -> None:
    self.db_path = Path(db_path or (get_data_dir() / "mesh.db"))
    self.db_path.parent.mkdir(parents=True, exist_ok=True)
    with closing(self._connect()) as conn, conn:
        # The old index was UNIQUE on the key alone. Left in place it
        # would keep enforcing the very collision this schema removes,
        # so it goes before the new one is created.
        conn.execute("DROP INDEX IF EXISTS mesh_commands_idem_idx")
        conn.executescript(_SCHEMA)
        self._ensure_columns(conn)
        conn.commit()

Methods:

enqueue

enqueue(
    command: RemoteCommand, *, status: str = "PENDING"
) -> dict

Record a command before any attempt to deliver it.

Written FIRST, on purpose: a command that left the machine without a row behind it is a command nobody can tell you about afterwards. This is the transactional-outbox rule of spec §16 applied to commands.

Source code in src/diapason/mesh/queue.py

def enqueue(self, command: RemoteCommand, *, status: str = "PENDING") -> dict:
    """Record a command before any attempt to deliver it.

    Written FIRST, on purpose: a command that left the machine without a
    row behind it is a command nobody can tell you about afterwards. This
    is the transactional-outbox rule of spec §16 applied to commands.
    """
    stamp = now_ms()
    with closing(self._connect()) as conn, conn:
        existing = conn.execute(
            "SELECT * FROM mesh_commands "
            "WHERE origin_device_id=? AND idempotency_key=?",
            (command.origin_device_id, command.idempotency_key),
        ).fetchone()
        if existing is not None:
            # Same intent, already recorded — hand back what we know
            # instead of creating a second one (spec §45).
            return self._serialize(existing)
        conn.execute(
            """INSERT INTO mesh_commands
               (command_id, idempotency_key, origin_device_id,
                target_device_id, tool, envelope_json, status,
                created_at_ms, expires_at_ms, updated_at_ms, sealed)
               VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
            (
                command.command_id,
                command.idempotency_key,
                command.origin_device_id,
                command.target_device_id,
                command.tool,
                json.dumps(command.to_dict(), ensure_ascii=False),
                status,
                command.created_at_ms,
                command.expires_at_ms,
                stamp,
                # La colonne, PAS `tool` : celui-ci garde le verbe en
                # clair, pour que l'historique reste lisible au lieu
                # d'afficher « mesh.sealed » quarante fois.
                1 if command.est_scelle else 0,
            ),
        )
        conn.commit()
    return self.get(command.command_id)

expire_stale

expire_stale(*, now: int | None = None) -> int

Turn past-deadline queued commands into honest EXPIRED rows.

Without this a queue quietly accumulates commands the user believes are still coming.

Source code in src/diapason/mesh/queue.py

def expire_stale(self, *, now: int | None = None) -> int:
    """Turn past-deadline queued commands into honest EXPIRED rows.

    Without this a queue quietly accumulates commands the user believes
    are still coming.
    """
    stamp = now_ms() if now is None else now
    with closing(self._connect()) as conn, conn:
        cursor = conn.execute(
            """UPDATE mesh_commands
               SET status='EXPIRED', completed_at_ms=?, updated_at_ms=?,
                   user_message='La commande a expiré avant d''être livrée.'
               WHERE status IN ('PENDING','QUEUED','ACCEPTED','RUNNING')
                 AND expires_at_ms < ?""",
            (stamp, stamp, stamp),
        )
        conn.commit()
    return cursor.rowcount

find_by_idempotency

find_by_idempotency(
    key: str, *, origin_device_id: str
) -> dict | None

The command a given sender already sent under this key.

Scoped by sender because the key is the SENDER's word for "the same intent". Two devices choosing the same string mean two different intents, and conflating them let a peer speak about our rows.

Source code in src/diapason/mesh/queue.py

def find_by_idempotency(self, key: str, *, origin_device_id: str) -> dict | None:
    """The command a given sender already sent under this key.

    Scoped by sender because the key is the SENDER's word for "the same
    intent". Two devices choosing the same string mean two different
    intents, and conflating them let a peer speak about our rows.
    """
    with closing(self._connect()) as conn:
        row = conn.execute(
            "SELECT * FROM mesh_commands "
            "WHERE origin_device_id=? AND idempotency_key=?",
            (origin_device_id, key),
        ).fetchone()
    return None if row is None else self._serialize(row)

pending_for

pending_for(
    target_device_id: str, *, limit: int = 50
) -> list[dict]

Queued commands still worth delivering to this device.

Source code in src/diapason/mesh/queue.py

def pending_for(self, target_device_id: str, *, limit: int = 50) -> list[dict]:
    """Queued commands still worth delivering to this device."""
    stamp = now_ms()
    with closing(self._connect()) as conn:
        rows = conn.execute(
            """SELECT * FROM mesh_commands
               WHERE target_device_id=? AND status IN ('PENDING','QUEUED')
                 AND expires_at_ms >= ?
               ORDER BY created_at_ms LIMIT ?""",
            (target_device_id, stamp, limit),
        ).fetchall()
    return [self._serialize(row) for row in rows]

envelope_of

envelope_of(command_id: str) -> RemoteCommand | None

The signed command as it was recorded, ready to travel again.

Re-sent verbatim rather than rebuilt: the signature covers the original bytes, so a command re-signed with a fresh timestamp would be a different command, and the receiver's replay protection could no longer tell a retry from a duplicate.

Source code in src/diapason/mesh/queue.py

def envelope_of(self, command_id: str) -> RemoteCommand | None:
    """The signed command as it was recorded, ready to travel again.

    Re-sent verbatim rather than rebuilt: the signature covers the
    original bytes, so a command re-signed with a fresh timestamp would
    be a *different* command, and the receiver's replay protection could
    no longer tell a retry from a duplicate.
    """
    with closing(self._connect()) as conn:
        row = conn.execute(
            "SELECT envelope_json FROM mesh_commands WHERE command_id=?",
            (command_id,),
        ).fetchone()
    if row is None:
        return None
    try:
        return RemoteCommand.from_dict(json.loads(row["envelope_json"]))
    except (TypeError, ValueError, KeyError):
        logger.warning("enveloppe illisible pour %s", command_id)
        return None

pending_envelopes_for

pending_envelopes_for(
    target_device_id: str, *, limit: int = 50
) -> list[dict]

The same queue, as signed envelopes a device can verify itself.

Separate from pending_for because the envelope is only ever wanted by the one caller that hands commands to a polling device. Putting it in every serialisation would push signatures through the command history and the UI, which have no use for them.

Source code in src/diapason/mesh/queue.py

def pending_envelopes_for(
    self, target_device_id: str, *, limit: int = 50
) -> list[dict]:
    """The same queue, as signed envelopes a device can verify itself.

    Separate from ``pending_for`` because the envelope is only ever wanted
    by the one caller that hands commands to a polling device. Putting it
    in every serialisation would push signatures through the command
    history and the UI, which have no use for them.
    """
    stamp = now_ms()
    with closing(self._connect()) as conn:
        rows = conn.execute(
            """SELECT command_id, envelope_json FROM mesh_commands
               WHERE target_device_id=? AND status IN ('PENDING','QUEUED')
                 AND expires_at_ms >= ?
               ORDER BY created_at_ms LIMIT ?""",
            (target_device_id, stamp, limit),
        ).fetchall()
    out: list[dict] = []
    for row in rows:
        try:
            envelope = json.loads(row["envelope_json"])
        except (TypeError, ValueError):
            # A row we cannot parse is one we cannot honestly deliver.
            # Skipping keeps the poll working for every other command.
            logger.warning("enveloppe illisible pour %s", row["command_id"])
            continue
        out.append({"commandId": row["command_id"], "envelope": envelope})
    return out

history

history(*, limit: int = 50) -> list[dict]

Recent commands, newest first — the §43 command history.

Source code in src/diapason/mesh/queue.py

def history(self, *, limit: int = 50) -> list[dict]:
    """Recent commands, newest first — the §43 command history."""
    with closing(self._connect()) as conn:
        rows = conn.execute(
            "SELECT * FROM mesh_commands ORDER BY created_at_ms DESC LIMIT ?",
            (limit,),
        ).fetchall()
    return [self._serialize(row) for row in rows]

Functions: