893 lines
30 KiB
Python
893 lines
30 KiB
Python
import asyncio
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import aiosqlite
|
|
import pytest
|
|
|
|
from haiku.rag.ingester.queue.migrations import apply_migrations, open_queue
|
|
from haiku.rag.ingester.queue.models import JobOp, JobStatus, SyncRow
|
|
from haiku.rag.ingester.queue.repository import JobRepo, SyncStateRepo
|
|
|
|
|
|
@pytest.fixture
|
|
async def conn(tmp_path):
|
|
path = tmp_path / "queue.db"
|
|
connection = await aiosqlite.connect(str(path))
|
|
connection.row_factory = aiosqlite.Row
|
|
await apply_migrations(connection)
|
|
yield connection
|
|
await connection.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def queue_lock():
|
|
return asyncio.Lock()
|
|
|
|
|
|
@pytest.fixture
|
|
def jobs(conn, queue_lock):
|
|
return JobRepo(conn, lock=queue_lock)
|
|
|
|
|
|
@pytest.fixture
|
|
def sync(conn, queue_lock):
|
|
return SyncStateRepo(conn, lock=queue_lock)
|
|
|
|
|
|
# --- migrations / schema ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_migrations_sets_schema_version(conn):
|
|
cursor = await conn.execute("SELECT version FROM schema_version")
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
assert row["version"] >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_migrations_is_idempotent(conn):
|
|
await apply_migrations(conn)
|
|
await apply_migrations(conn)
|
|
cursor = await conn.execute("SELECT COUNT(*) AS n FROM schema_version")
|
|
row = await cursor.fetchone()
|
|
assert row["n"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_open_queue_creates_file_and_schema(tmp_path):
|
|
path = tmp_path / "subdir" / "queue.db"
|
|
connection = await open_queue(path)
|
|
try:
|
|
assert path.exists()
|
|
cursor = await connection.execute("SELECT version FROM schema_version")
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
finally:
|
|
await connection.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dlq_view_exposes_dead_jobs(conn, jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_dead(job.id, "permanent", "w")
|
|
|
|
cursor = await conn.execute("SELECT id FROM dlq")
|
|
rows = await cursor.fetchall()
|
|
assert [r["id"] for r in rows] == [job.id]
|
|
|
|
|
|
# --- enqueue ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueue_creates_queued_job(jobs):
|
|
job = await jobs.enqueue(
|
|
"fs:/tmp",
|
|
"file:///tmp/a.md",
|
|
JobOp.UPSERT,
|
|
revision="abc",
|
|
content_hash="md5_value",
|
|
extra={"key": "value"},
|
|
)
|
|
assert job is not None
|
|
assert job.status is JobStatus.QUEUED
|
|
assert job.source_id == "fs:/tmp"
|
|
assert job.uri == "file:///tmp/a.md"
|
|
assert job.op is JobOp.UPSERT
|
|
assert job.revision == "abc"
|
|
assert job.content_hash == "md5_value"
|
|
assert job.attempts == 0
|
|
assert job.extra == {"key": "value"}
|
|
assert job.enqueued_at.tzinfo is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueue_returns_none_on_live_conflict(jobs):
|
|
first = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
second = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert first is not None
|
|
assert second is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueue_after_dead_succeeds(jobs):
|
|
first = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_dead(first.id, "error", "w")
|
|
second = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert second is not None
|
|
assert second.id != first.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueue_after_succeeded_succeeds(jobs):
|
|
await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_succeeded(claimed.id, "w")
|
|
second = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert second is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_has_pending_returns_false_on_empty_queue(jobs):
|
|
assert await jobs.has_pending("src") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_has_pending_true_for_queued_job(jobs):
|
|
await jobs.enqueue("src", "u", JobOp.UPSERT)
|
|
assert await jobs.has_pending("src") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_has_pending_true_for_claimed_job(jobs):
|
|
await jobs.enqueue("src", "u", JobOp.UPSERT)
|
|
await jobs.claim_next("w")
|
|
assert await jobs.has_pending("src") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_has_pending_false_for_terminal_states(jobs):
|
|
await jobs.enqueue("src", "u-ok", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_succeeded(claimed.id, "w")
|
|
|
|
await jobs.enqueue("src", "u-bad", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_dead(claimed.id, "permanent", "w")
|
|
|
|
assert await jobs.has_pending("src") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_has_pending_is_per_source(jobs):
|
|
"""A backed-up source must not block an idle one."""
|
|
await jobs.enqueue("busy", "u", JobOp.UPSERT)
|
|
assert await jobs.has_pending("busy") is True
|
|
assert await jobs.has_pending("idle") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueue_drops_delete_when_upsert_is_live(jobs):
|
|
"""Stops a DELETE worker from removing a document a sibling UPSERT
|
|
just ingested."""
|
|
upsert = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
delete = await jobs.enqueue("s", "u", JobOp.DELETE)
|
|
assert upsert is not None
|
|
assert delete is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueue_drops_upsert_when_delete_is_live(jobs):
|
|
"""Symmetric: live DELETE for a URI blocks a fresh UPSERT for the
|
|
same URI. The next sweep after DELETE completes will re-emit the
|
|
UPSERT if the file is back."""
|
|
delete = await jobs.enqueue("s", "u", JobOp.DELETE)
|
|
upsert = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert delete is not None
|
|
assert upsert is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enqueue_different_uris_independent_of_op(jobs):
|
|
"""Uniqueness is per-(source_id, uri), not per-(source_id, uri, op).
|
|
Different URIs can be queued regardless of which op each one is."""
|
|
a = await jobs.enqueue("s", "u-a", JobOp.UPSERT)
|
|
b = await jobs.enqueue("s", "u-b", JobOp.DELETE)
|
|
assert a is not None
|
|
assert b is not None
|
|
|
|
|
|
# --- claim_next ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_returns_none_when_empty(jobs):
|
|
assert await jobs.claim_next("w") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_increments_attempts_and_records_worker(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert job.attempts == 0
|
|
claimed = await jobs.claim_next("worker-1")
|
|
assert claimed is not None
|
|
assert claimed.id == job.id
|
|
assert claimed.status is JobStatus.CLAIMED
|
|
assert claimed.attempts == 1
|
|
assert claimed.claimed_by == "worker-1"
|
|
assert claimed.claimed_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_returns_oldest_first(jobs):
|
|
j1 = await jobs.enqueue("s", "u1", JobOp.UPSERT)
|
|
j2 = await jobs.enqueue("s", "u2", JobOp.UPSERT)
|
|
first = await jobs.claim_next("w")
|
|
second = await jobs.claim_next("w")
|
|
assert first is not None
|
|
assert second is not None
|
|
assert first.id == j1.id
|
|
assert second.id == j2.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_skips_future_scheduled(conn, jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
# Push scheduled_at into the future.
|
|
future = (datetime.now(UTC) + timedelta(seconds=60)).isoformat()
|
|
await conn.execute(
|
|
"UPDATE jobs SET scheduled_at = ? WHERE id = ?", (future, job.id)
|
|
)
|
|
await conn.commit()
|
|
assert await jobs.claim_next("w") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_next_atomic_under_concurrency(jobs):
|
|
enqueued = []
|
|
for i in range(5):
|
|
j = await jobs.enqueue("s", f"u{i}", JobOp.UPSERT)
|
|
assert j is not None
|
|
enqueued.append(j)
|
|
|
|
results = await asyncio.gather(*(jobs.claim_next(f"w{i}") for i in range(10)))
|
|
claimed = [r for r in results if r is not None]
|
|
assert len(claimed) == 5
|
|
assert len({c.id for c in claimed}) == 5
|
|
assert {c.id for c in claimed} == {j.id for j in enqueued}
|
|
|
|
|
|
# --- terminal transitions ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_succeeded(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_succeeded(claimed.id, "w")
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.SUCCEEDED
|
|
assert refreshed.completed_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_dead_records_error(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_dead(claimed.id, "permanent failure", "w")
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.DEAD
|
|
assert refreshed.last_error == "permanent failure"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_succeeded_with_claimed_by_guard_skips_when_resurrected(jobs):
|
|
"""Reaper race: A claims, reaper resets, B re-claims, A finishes. A's
|
|
mark_succeeded must be a no-op so B's in-flight work isn't clobbered."""
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
a = await jobs.claim_next("worker-A")
|
|
assert a is not None
|
|
# Reaper resets A's stale claim.
|
|
await jobs.reap_stale(claim_timeout_seconds=0)
|
|
# B picks the job up.
|
|
b = await jobs.claim_next("worker-B")
|
|
assert b is not None and b.id == job.id and b.claimed_by == "worker-B"
|
|
# A finally finishes and tries to mark succeeded — guarded, no-op.
|
|
assert await jobs.mark_succeeded(job.id, "worker-A") is False
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.CLAIMED
|
|
assert refreshed.claimed_by == "worker-B"
|
|
# B's own mark_succeeded does land.
|
|
assert await jobs.mark_succeeded(job.id, "worker-B") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_dead_with_claimed_by_guard_skips_when_resurrected(jobs):
|
|
"""Same guard semantics as mark_succeeded for the dead transition."""
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
a = await jobs.claim_next("worker-A")
|
|
assert a is not None
|
|
await jobs.reap_stale(claim_timeout_seconds=0)
|
|
b = await jobs.claim_next("worker-B")
|
|
assert b is not None and b.claimed_by == "worker-B"
|
|
assert await jobs.mark_dead(job.id, "boom", "worker-A") is False
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.CLAIMED
|
|
assert refreshed.last_error is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reschedule_with_claimed_by_guard_skips_when_resurrected(jobs):
|
|
"""Worker A times out → reaper resets → worker B re-claims → A surfaces
|
|
with a TransientError and calls reschedule. The guard turns A's call
|
|
into a no-op so B's claim isn't stripped back to `queued`."""
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
a = await jobs.claim_next("worker-A")
|
|
assert a is not None
|
|
await jobs.reap_stale(claim_timeout_seconds=0)
|
|
b = await jobs.claim_next("worker-B")
|
|
assert b is not None and b.claimed_by == "worker-B"
|
|
assert await jobs.reschedule(job.id, 30.0, "transient", "worker-A") is False
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.CLAIMED
|
|
assert refreshed.claimed_by == "worker-B"
|
|
assert refreshed.last_error is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_release_if_claimed_with_claimed_by_guard_skips_when_resurrected(jobs):
|
|
"""Cancel-cleanup path: worker A is cancelled while reaper-resurrected.
|
|
release_if_claimed must not strip worker B's fresh claim."""
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
a = await jobs.claim_next("worker-A")
|
|
assert a is not None
|
|
await jobs.reap_stale(claim_timeout_seconds=0)
|
|
b = await jobs.claim_next("worker-B")
|
|
assert b is not None and b.claimed_by == "worker-B"
|
|
assert await jobs.release_if_claimed(job.id, "worker-A") is False
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.CLAIMED
|
|
assert refreshed.claimed_by == "worker-B"
|
|
|
|
|
|
# --- reschedule + retry ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reschedule_pushes_scheduled_at_into_future(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.reschedule(claimed.id, 30.0, "transient", "w")
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.QUEUED
|
|
assert refreshed.claimed_at is None
|
|
assert refreshed.claimed_by is None
|
|
assert refreshed.last_error == "transient"
|
|
assert refreshed.scheduled_at > datetime.now(UTC)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reschedule_then_claim_skips_until_due(conn, jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.reschedule(claimed.id, 60.0, "transient", "w")
|
|
assert await jobs.claim_next("w") is None
|
|
|
|
# Backdate scheduled_at to simulate the delay elapsing.
|
|
past = datetime.now(UTC).isoformat()
|
|
await conn.execute("UPDATE jobs SET scheduled_at = ? WHERE id = ?", (past, job.id))
|
|
await conn.commit()
|
|
assert await jobs.claim_next("w") is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_revives_dead_job(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_dead(claimed.id, "boom", "w")
|
|
|
|
revived = await jobs.retry(job.id)
|
|
assert revived.status is JobStatus.QUEUED
|
|
assert revived.attempts == 0
|
|
assert revived.last_error is None
|
|
assert revived.claimed_at is None
|
|
assert revived.claimed_by is None
|
|
assert revived.completed_at is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_unknown_raises(jobs):
|
|
with pytest.raises(KeyError):
|
|
await jobs.retry("not-a-real-id")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_refuses_claimed_job(jobs):
|
|
"""Resetting a `claimed` row would race with the worker still
|
|
processing it: claim_next would re-claim and a second worker would
|
|
process the same URI."""
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
assert claimed.status is JobStatus.CLAIMED
|
|
|
|
with pytest.raises(KeyError):
|
|
await jobs.retry(job.id)
|
|
|
|
# Row is untouched.
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.CLAIMED
|
|
assert refreshed.attempts == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_refuses_succeeded_job(jobs):
|
|
"""Succeeded rows should be re-ingested through the UPSERT path, not
|
|
re-run from the queue."""
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_succeeded(claimed.id, "w")
|
|
|
|
with pytest.raises(KeyError):
|
|
await jobs.retry(job.id)
|
|
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.SUCCEEDED
|
|
|
|
|
|
# --- cancel ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_queued_removes_row(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert await jobs.cancel(job.id) is True
|
|
assert await jobs.get_job(job.id) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_succeeded_returns_false(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_succeeded(claimed.id, "w")
|
|
assert await jobs.cancel(job.id) is False
|
|
|
|
|
|
# --- reap_stale ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reap_stale_resets_old_claims(conn, jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
assert claimed.attempts == 1
|
|
|
|
# Backdate claimed_at to simulate a crashed worker.
|
|
long_ago = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
|
|
await conn.execute(
|
|
"UPDATE jobs SET claimed_at = ? WHERE id = ?", (long_ago, job.id)
|
|
)
|
|
await conn.commit()
|
|
|
|
reset = await jobs.reap_stale(claim_timeout_seconds=60)
|
|
assert reset == 1
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.QUEUED
|
|
assert refreshed.claimed_at is None
|
|
assert refreshed.claimed_by is None
|
|
# claim_next incremented to 1; reap undoes that since a crashed worker
|
|
# isn't a consumed attempt — matches release_if_claimed semantics.
|
|
assert refreshed.attempts == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reap_stale_leaves_fresh_claims_alone(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
|
|
reset = await jobs.reap_stale(claim_timeout_seconds=3600)
|
|
assert reset == 0
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.CLAIMED
|
|
|
|
|
|
# --- prune_dead ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prune_dead_removes_matching_dead_rows(jobs):
|
|
"""A dead UPSERT becomes stale once a sibling DELETE has resolved the URI;
|
|
prune_dead() removes it so the DLQ stops showing resolved entries."""
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert job is not None
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_dead(claimed.id, "boom", "w")
|
|
|
|
pruned = await jobs.prune_dead("s", "u")
|
|
assert pruned == 1
|
|
assert await jobs.get_job(job.id) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prune_dead_leaves_non_dead_rows_alone(jobs):
|
|
"""Queued/claimed/succeeded rows for the same (source, uri) are not
|
|
touched — only `dead` is purged."""
|
|
queued = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert queued is not None
|
|
|
|
pruned = await jobs.prune_dead("s", "u")
|
|
assert pruned == 0
|
|
refreshed = await jobs.get_job(queued.id)
|
|
assert refreshed is not None and refreshed.status is JobStatus.QUEUED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prune_dead_scoped_to_matching_uri(jobs):
|
|
"""Dead rows for other URIs (and other sources) survive."""
|
|
j1 = await jobs.enqueue("s", "u1", JobOp.UPSERT)
|
|
assert j1 is not None
|
|
await jobs.mark_dead((await jobs.claim_next("w")).id, "err", "w")
|
|
|
|
j2 = await jobs.enqueue("s", "u2", JobOp.UPSERT)
|
|
assert j2 is not None
|
|
await jobs.mark_dead((await jobs.claim_next("w")).id, "err", "w")
|
|
|
|
pruned = await jobs.prune_dead("s", "u1")
|
|
assert pruned == 1
|
|
assert await jobs.get_job(j1.id) is None
|
|
assert await jobs.get_job(j2.id) is not None
|
|
|
|
|
|
# --- release_if_claimed ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_release_if_claimed_resets_claimed_job_and_decrements_attempts(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert job is not None
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
assert claimed.attempts == 1
|
|
|
|
released = await jobs.release_if_claimed(job.id, "w")
|
|
assert released is True
|
|
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.QUEUED
|
|
assert refreshed.claimed_at is None
|
|
assert refreshed.claimed_by is None
|
|
assert refreshed.attempts == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_release_if_claimed_noop_on_already_queued(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert job is not None
|
|
|
|
released = await jobs.release_if_claimed(job.id, "w")
|
|
assert released is False
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.QUEUED
|
|
assert refreshed.attempts == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_release_if_claimed_noop_on_succeeded(jobs):
|
|
job = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert job is not None
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_succeeded(job.id, "w")
|
|
|
|
released = await jobs.release_if_claimed(job.id, "w")
|
|
assert released is False
|
|
refreshed = await jobs.get_job(job.id)
|
|
assert refreshed is not None
|
|
assert refreshed.status is JobStatus.SUCCEEDED
|
|
|
|
|
|
# --- list / counts ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_jobs_with_filters(jobs):
|
|
j1 = await jobs.enqueue("s1", "u1", JobOp.UPSERT)
|
|
j2 = await jobs.enqueue("s2", "u2", JobOp.UPSERT)
|
|
j3 = await jobs.enqueue("s1", "u3", JobOp.UPSERT)
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_succeeded(claimed.id, "w")
|
|
|
|
all_jobs = await jobs.list_jobs()
|
|
assert len(all_jobs) == 3
|
|
|
|
by_source = await jobs.list_jobs(source_id="s1")
|
|
assert {j.id for j in by_source} == {j1.id, j3.id}
|
|
|
|
by_status = await jobs.list_jobs(status=JobStatus.QUEUED)
|
|
assert {j.id for j in by_status} == {j2.id, j3.id}
|
|
|
|
by_uri = await jobs.list_jobs(uri="u3")
|
|
assert {j.id for j in by_uri} == {j3.id}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_counts_by_status(jobs):
|
|
await jobs.enqueue("s", "u1", JobOp.UPSERT)
|
|
j2 = await jobs.enqueue("s", "u2", JobOp.UPSERT)
|
|
j3 = await jobs.enqueue("s", "u3", JobOp.UPSERT)
|
|
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None
|
|
await jobs.mark_succeeded(claimed.id, "w")
|
|
claimed_j2 = await jobs.claim_next("w")
|
|
assert claimed_j2 is not None and claimed_j2.id == j2.id
|
|
await jobs.mark_dead(j2.id, "err", "w")
|
|
|
|
counts = await jobs.counts_by_status()
|
|
assert counts == {"queued": 1, "succeeded": 1, "dead": 1}
|
|
assert j3.id # silence unused
|
|
|
|
|
|
# --- stats ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_count_succeeded_since_only_includes_recent(jobs, conn):
|
|
await jobs.enqueue("s", "old", JobOp.UPSERT)
|
|
await jobs.enqueue("s", "new", JobOp.UPSERT)
|
|
|
|
old_claim = await jobs.claim_next("w")
|
|
assert old_claim is not None
|
|
await jobs.mark_succeeded(old_claim.id, "w")
|
|
long_ago = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
|
|
await conn.execute(
|
|
"UPDATE jobs SET completed_at = ? WHERE id = ?", (long_ago, old_claim.id)
|
|
)
|
|
await conn.commit()
|
|
|
|
new_claim = await jobs.claim_next("w")
|
|
assert new_claim is not None
|
|
await jobs.mark_succeeded(new_claim.id, "w")
|
|
|
|
assert await jobs.count_succeeded_since(60) == 1
|
|
assert await jobs.count_succeeded_since(86400) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_oldest_queued_age_seconds_none_when_empty(jobs):
|
|
assert await jobs.oldest_queued_age_seconds() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_oldest_queued_age_seconds_returns_oldest(jobs, conn):
|
|
old = await jobs.enqueue("s", "old", JobOp.UPSERT)
|
|
await jobs.enqueue("s", "new", JobOp.UPSERT)
|
|
backdate = (datetime.now(UTC) - timedelta(seconds=120)).isoformat()
|
|
assert old is not None
|
|
await conn.execute(
|
|
"UPDATE jobs SET scheduled_at = ? WHERE id = ?", (backdate, old.id)
|
|
)
|
|
await conn.commit()
|
|
|
|
age = await jobs.oldest_queued_age_seconds()
|
|
assert age is not None
|
|
assert 119 <= age <= 125
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_oldest_queued_age_seconds_ignores_future_scheduled(jobs, conn):
|
|
"""A job whose scheduled_at is in the future (e.g. after a backoff
|
|
reschedule) isn't ready to run, so it shouldn't count toward backlog age."""
|
|
j = await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert j is not None
|
|
future = (datetime.now(UTC) + timedelta(seconds=600)).isoformat()
|
|
await conn.execute("UPDATE jobs SET scheduled_at = ? WHERE id = ?", (future, j.id))
|
|
await conn.commit()
|
|
assert await jobs.oldest_queued_age_seconds() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_counts_by_source_groups_correctly(jobs):
|
|
# Enqueue s2 first so claim_next picks it up before the s1 rows; then
|
|
# mark_dead routes through the production claim→terminal transition.
|
|
j3 = await jobs.enqueue("s2", "u3", JobOp.UPSERT)
|
|
assert j3 is not None
|
|
claimed = await jobs.claim_next("w")
|
|
assert claimed is not None and claimed.id == j3.id
|
|
await jobs.mark_dead(j3.id, "boom", "w")
|
|
await jobs.enqueue("s1", "u1", JobOp.UPSERT)
|
|
await jobs.enqueue("s1", "u2", JobOp.UPSERT)
|
|
|
|
assert await jobs.counts_by_source("queued") == {"s1": 2}
|
|
assert await jobs.counts_by_source("dead") == {"s2": 1}
|
|
assert await jobs.counts_by_source("queued", "claimed") == {"s1": 2}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_counts_by_source_no_statuses_returns_empty(jobs):
|
|
await jobs.enqueue("s", "u", JobOp.UPSERT)
|
|
assert await jobs.counts_by_source() == {}
|
|
|
|
|
|
# --- cross-repo lock sharing ---
|
|
|
|
|
|
def test_repos_share_lock_when_constructed_with_one(conn):
|
|
"""JobRepo and SyncStateRepo wrap one shared aiosqlite.Connection in
|
|
production. They must accept and share a single asyncio.Lock so cross-
|
|
repo calls serialize at the cursor/commit boundary — otherwise a
|
|
SyncStateRepo.upsert cursor open while JobRepo.mark_succeeded tries
|
|
to commit would trip SQLite's 'SQL statements in progress' error."""
|
|
shared = asyncio.Lock()
|
|
j = JobRepo(conn, lock=shared)
|
|
s = SyncStateRepo(conn, lock=shared)
|
|
assert j._lock is s._lock is shared
|
|
|
|
|
|
def test_repos_default_to_independent_locks_when_used_alone(conn):
|
|
"""Backward-compat: a single-repo caller can still construct without
|
|
passing a lock and each repo creates its own."""
|
|
j = JobRepo(conn)
|
|
s = SyncStateRepo(conn)
|
|
assert j._lock is not s._lock
|
|
|
|
|
|
# --- sync state ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_get_revision_snapshot_empty(sync):
|
|
assert await sync.get_revision_snapshot("unknown") == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_upsert_and_get(sync):
|
|
await sync.upsert("s", "u1", revision="abc", content_hash="m1")
|
|
await sync.upsert("s", "u2", revision="def", content_hash="m2")
|
|
assert await sync.get_revision_snapshot("s") == {"u1": "abc", "u2": "def"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_upsert_overwrites(sync):
|
|
await sync.upsert("s", "u1", revision="abc", content_hash="m1")
|
|
await sync.upsert("s", "u1", revision="def", content_hash="m2", ingested=True)
|
|
assert await sync.get_revision_snapshot("s") == {"u1": "def"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_delete_removes_entry(sync):
|
|
await sync.upsert("s", "u1", revision="abc", content_hash="m1")
|
|
await sync.delete("s", "u1")
|
|
assert await sync.get_revision_snapshot("s") == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_snapshot_scoped_per_source(sync):
|
|
await sync.upsert("s1", "u", revision="abc", content_hash="m")
|
|
await sync.upsert("s2", "u", revision="def", content_hash="m")
|
|
assert await sync.get_revision_snapshot("s1") == {"u": "abc"}
|
|
assert await sync.get_revision_snapshot("s2") == {"u": "def"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_revision_snapshot_excludes_null_revision_rows(sync):
|
|
"""get_revision_snapshot returns only rows with a stored revision —
|
|
sources compare against this for UPSERT/UNCHANGED. Rows without a
|
|
revision (HTTP without ETag, worker DLQ'd before completion) are
|
|
visible via list_known_uris instead."""
|
|
await sync.upsert("s", "u", revision=None, content_hash=None)
|
|
assert await sync.get_revision_snapshot("s") == {}
|
|
assert await sync.list_known_uris("s") == {"u"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_upsert_preserves_revision_when_none(sync):
|
|
"""upsert(revision=None) leaves an existing revision in place."""
|
|
await sync.upsert("s", "u", revision="v1", content_hash="hash-v1")
|
|
await sync.upsert("s", "u", revision=None, content_hash=None)
|
|
row = await sync.get_row("s", "u")
|
|
assert row is not None
|
|
assert row.revision == "v1"
|
|
assert row.content_hash == "hash-v1"
|
|
assert await sync.get_revision_snapshot("s") == {"u": "v1"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_upsert_replaces_revision_when_provided(sync):
|
|
"""upsert with a non-None revision overwrites the existing one."""
|
|
await sync.upsert("s", "u", revision="v1", content_hash="hash-v1")
|
|
await sync.upsert("s", "u", revision="v2", content_hash="hash-v2", ingested=True)
|
|
row = await sync.get_row("s", "u")
|
|
assert row is not None
|
|
assert row.revision == "v2"
|
|
assert row.content_hash == "hash-v2"
|
|
assert row.last_ingested_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_batch_upsert_inserts_multiple_rows(sync):
|
|
"""batch_upsert writes many rows in a single transaction."""
|
|
await sync.batch_upsert(
|
|
[
|
|
SyncRow("s", "u1", "rev1", "hash1", False),
|
|
SyncRow("s", "u2", "rev2", "hash2", False),
|
|
SyncRow("s", "u3", "rev3", None, True),
|
|
]
|
|
)
|
|
assert await sync.get_revision_snapshot("s") == {
|
|
"u1": "rev1",
|
|
"u2": "rev2",
|
|
"u3": "rev3",
|
|
}
|
|
row = await sync.get_row("s", "u3")
|
|
assert row is not None
|
|
assert row.last_ingested_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_batch_upsert_updates_existing(sync):
|
|
"""batch_upsert applies ON CONFLICT update semantics like upsert()."""
|
|
await sync.upsert("s", "u1", revision="old", content_hash="old-hash")
|
|
await sync.batch_upsert(
|
|
[
|
|
SyncRow("s", "u1", "new", "new-hash", False),
|
|
]
|
|
)
|
|
row = await sync.get_row("s", "u1")
|
|
assert row is not None
|
|
assert row.revision == "new"
|
|
assert row.content_hash == "new-hash"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_batch_upsert_preserves_revision_when_none(sync):
|
|
"""batch_upsert with revision=None leaves existing revision in place."""
|
|
await sync.upsert("s", "u1", revision="keep", content_hash="keep-hash")
|
|
await sync.batch_upsert(
|
|
[
|
|
SyncRow("s", "u1", None, None, False),
|
|
]
|
|
)
|
|
row = await sync.get_row("s", "u1")
|
|
assert row is not None
|
|
assert row.revision == "keep"
|
|
assert row.content_hash == "keep-hash"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_state_batch_upsert_empty_is_noop(sync):
|
|
"""batch_upsert with an empty list does nothing."""
|
|
await sync.batch_upsert([])
|
|
assert await sync.get_revision_snapshot("s") == {}
|