feat(subscriptions): live posts-processed progress on backfill/recovery — #704 step 2
The running badge only showed the chunk counter; now it shows posts walked — real walk progress. The ingester already reports posts_processed per chunk (step 1); the backfill lifecycle accumulates it into config_overrides._backfill_posts across chunks. SourceRecord exposes backfill_posts; start_backfill/start_recovery clear it (fresh walk); stop clears it too. SourceRow/SourceCard badge renders "Recovering · 45 posts" (falls back to "(N)" chunks before any posts are counted). Per-chunk accumulation (no mid-walk DB write) — simple and race-free; a small over-count from each chunk re-walking its resumed page is fine for a progress indicator. Tests: lifecycle accumulates posts_processed across chunks; start clears a prior _backfill_posts and the record exposes it. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -515,6 +515,13 @@ class DownloadService:
|
|||||||
new_overrides = dict(src.config_overrides or {})
|
new_overrides = dict(src.config_overrides or {})
|
||||||
chunks = int(new_overrides.get("_backfill_chunks", 0)) + 1
|
chunks = int(new_overrides.get("_backfill_chunks", 0)) + 1
|
||||||
new_overrides["_backfill_chunks"] = chunks
|
new_overrides["_backfill_chunks"] = chunks
|
||||||
|
# plan #704 (#5): accumulate posts-processed across chunks so the badge
|
||||||
|
# shows live walk progress, not just the chunk count. Cleared on
|
||||||
|
# (re)start. (Slight over-count: each chunk re-walks the resumed page —
|
||||||
|
# fine for a progress indicator.)
|
||||||
|
new_overrides["_backfill_posts"] = int(
|
||||||
|
new_overrides.get("_backfill_posts", 0)
|
||||||
|
) + int(getattr(dl_result, "posts_processed", 0) or 0)
|
||||||
|
|
||||||
completed = (
|
completed = (
|
||||||
dl_result.success
|
dl_result.success
|
||||||
|
|||||||
@@ -70,6 +70,9 @@ class SourceRecord:
|
|||||||
# plan #697: a running deep-walk that bypasses the Patreon seen-ledger
|
# plan #697: a running deep-walk that bypasses the Patreon seen-ledger
|
||||||
# (recovery) vs. a normal backfill. Lets the badge label it "Recovering".
|
# (recovery) vs. a normal backfill. Lets the badge label it "Recovering".
|
||||||
backfill_bypass_seen: bool
|
backfill_bypass_seen: bool
|
||||||
|
# plan #704: cumulative posts processed across the walk's chunks — live
|
||||||
|
# progress for the badge.
|
||||||
|
backfill_posts: int
|
||||||
|
|
||||||
def to_dict(self) -> dict:
|
def to_dict(self) -> dict:
|
||||||
return {
|
return {
|
||||||
@@ -91,6 +94,7 @@ class SourceRecord:
|
|||||||
"backfill_state": self.backfill_state,
|
"backfill_state": self.backfill_state,
|
||||||
"backfill_chunks": self.backfill_chunks,
|
"backfill_chunks": self.backfill_chunks,
|
||||||
"backfill_bypass_seen": self.backfill_bypass_seen,
|
"backfill_bypass_seen": self.backfill_bypass_seen,
|
||||||
|
"backfill_posts": self.backfill_posts,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -167,6 +171,7 @@ class SourceService:
|
|||||||
backfill_state=co.get("_backfill_state"),
|
backfill_state=co.get("_backfill_state"),
|
||||||
backfill_chunks=int(co.get("_backfill_chunks", 0)),
|
backfill_chunks=int(co.get("_backfill_chunks", 0)),
|
||||||
backfill_bypass_seen=bool(co.get("_backfill_bypass_seen")),
|
backfill_bypass_seen=bool(co.get("_backfill_bypass_seen")),
|
||||||
|
backfill_posts=int(co.get("_backfill_posts", 0)),
|
||||||
)
|
)
|
||||||
|
|
||||||
async def _row_to_record(self, source: Source) -> SourceRecord:
|
async def _row_to_record(self, source: Source) -> SourceRecord:
|
||||||
@@ -317,7 +322,8 @@ class SourceService:
|
|||||||
raise LookupError(f"source id={source_id} not found")
|
raise LookupError(f"source id={source_id} not found")
|
||||||
co = dict(source.config_overrides or {})
|
co = dict(source.config_overrides or {})
|
||||||
co["_backfill_state"] = "running"
|
co["_backfill_state"] = "running"
|
||||||
for k in ("_backfill_cursor", "_backfill_cursor_stalls", "_backfill_chunks"):
|
for k in ("_backfill_cursor", "_backfill_cursor_stalls", "_backfill_chunks",
|
||||||
|
"_backfill_posts"):
|
||||||
co.pop(k, None)
|
co.pop(k, None)
|
||||||
source.config_overrides = co
|
source.config_overrides = co
|
||||||
source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
|
source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
|
||||||
@@ -345,7 +351,8 @@ class SourceService:
|
|||||||
co = dict(source.config_overrides or {})
|
co = dict(source.config_overrides or {})
|
||||||
co["_backfill_state"] = "running"
|
co["_backfill_state"] = "running"
|
||||||
co["_backfill_bypass_seen"] = True
|
co["_backfill_bypass_seen"] = True
|
||||||
for k in ("_backfill_cursor", "_backfill_cursor_stalls", "_backfill_chunks"):
|
for k in ("_backfill_cursor", "_backfill_cursor_stalls", "_backfill_chunks",
|
||||||
|
"_backfill_posts"):
|
||||||
co.pop(k, None)
|
co.pop(k, None)
|
||||||
source.config_overrides = co
|
source.config_overrides = co
|
||||||
source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
|
source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
|
||||||
@@ -362,7 +369,7 @@ class SourceService:
|
|||||||
raise LookupError(f"source id={source_id} not found")
|
raise LookupError(f"source id={source_id} not found")
|
||||||
co = dict(source.config_overrides or {})
|
co = dict(source.config_overrides or {})
|
||||||
for k in ("_backfill_state", "_backfill_cursor", "_backfill_cursor_stalls",
|
for k in ("_backfill_state", "_backfill_cursor", "_backfill_cursor_stalls",
|
||||||
"_backfill_chunks", "_backfill_bypass_seen"):
|
"_backfill_chunks", "_backfill_bypass_seen", "_backfill_posts"):
|
||||||
co.pop(k, None)
|
co.pop(k, None)
|
||||||
source.config_overrides = co
|
source.config_overrides = co
|
||||||
source.backfill_runs_remaining = 0
|
source.backfill_runs_remaining = 0
|
||||||
|
|||||||
@@ -28,7 +28,9 @@
|
|||||||
v-else-if="source.backfill_state === 'running'"
|
v-else-if="source.backfill_state === 'running'"
|
||||||
size="x-small" color="info" variant="tonal" label
|
size="x-small" color="info" variant="tonal" label
|
||||||
>{{ source.backfill_bypass_seen ? 'Recovering' : 'Backfilling'
|
>{{ source.backfill_bypass_seen ? 'Recovering' : 'Backfilling'
|
||||||
}}{{ source.backfill_chunks ? ` (${source.backfill_chunks})` : '' }}</v-chip>
|
}}{{ source.backfill_posts
|
||||||
|
? ` · ${source.backfill_posts} posts`
|
||||||
|
: (source.backfill_chunks ? ` (${source.backfill_chunks})` : '') }}</v-chip>
|
||||||
<v-chip
|
<v-chip
|
||||||
v-else-if="source.backfill_state === 'complete'"
|
v-else-if="source.backfill_state === 'complete'"
|
||||||
size="x-small" color="success" variant="tonal" label
|
size="x-small" color="success" variant="tonal" label
|
||||||
|
|||||||
@@ -35,7 +35,9 @@
|
|||||||
v-else-if="source.backfill_state === 'running'"
|
v-else-if="source.backfill_state === 'running'"
|
||||||
size="x-small" color="info" variant="tonal" label
|
size="x-small" color="info" variant="tonal" label
|
||||||
>{{ source.backfill_bypass_seen ? 'Recovering' : 'Backfilling'
|
>{{ source.backfill_bypass_seen ? 'Recovering' : 'Backfilling'
|
||||||
}}{{ source.backfill_chunks ? ` (${source.backfill_chunks})` : '' }}</v-chip>
|
}}{{ source.backfill_posts
|
||||||
|
? ` · ${source.backfill_posts} posts`
|
||||||
|
: (source.backfill_chunks ? ` (${source.backfill_chunks})` : '') }}</v-chip>
|
||||||
<v-chip
|
<v-chip
|
||||||
v-else-if="source.backfill_state === 'complete'"
|
v-else-if="source.backfill_state === 'complete'"
|
||||||
size="x-small" color="success" variant="tonal" label
|
size="x-small" color="success" variant="tonal" label
|
||||||
|
|||||||
@@ -648,6 +648,40 @@ async def test_backfill_stuck_guard_stalls_after_two_no_advance(
|
|||||||
assert "_backfill_cursor" not in co2
|
assert "_backfill_cursor" not in co2
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_backfill_accumulates_posts_processed(
|
||||||
|
db, db_sync, tmp_path, seed_artist_and_source,
|
||||||
|
):
|
||||||
|
"""plan #704 (#5): the lifecycle accumulates DownloadResult.posts_processed
|
||||||
|
into _backfill_posts across chunks for the live progress badge."""
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
from backend.app.services.download_service import DownloadService
|
||||||
|
|
||||||
|
_artist, source = seed_artist_and_source
|
||||||
|
source.config_overrides = {"_backfill_state": "running", "_backfill_posts": 5}
|
||||||
|
source.backfill_runs_remaining = 5
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
|
svc = DownloadService(
|
||||||
|
async_session=db, sync_session=db_sync,
|
||||||
|
gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
|
||||||
|
)
|
||||||
|
ctx = {
|
||||||
|
"source_id": source.id, "platform": "patreon",
|
||||||
|
"config_overrides": {"_backfill_state": "running", "_backfill_posts": 5},
|
||||||
|
"backfill_runs_remaining": 5,
|
||||||
|
}
|
||||||
|
await svc._apply_backfill_lifecycle(
|
||||||
|
ctx, _make_fake_dl_result(success=False, cursor="C1", posts_processed=10),
|
||||||
|
)
|
||||||
|
await db.commit()
|
||||||
|
co = (await db.execute(
|
||||||
|
select(Source.config_overrides).where(Source.id == source.id)
|
||||||
|
)).scalar_one()
|
||||||
|
assert co.get("_backfill_posts") == 15 # 5 (prior) + 10 (this chunk)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_backfill_timeout_chunk_reclassified_to_ok(
|
async def test_backfill_timeout_chunk_reclassified_to_ok(
|
||||||
db, db_sync, tmp_path, seed_artist_and_source,
|
db, db_sync, tmp_path, seed_artist_and_source,
|
||||||
|
|||||||
@@ -238,15 +238,25 @@ async def test_start_recovery_arms_bypass_flag(db):
|
|||||||
artist_id=artist.id, platform="patreon",
|
artist_id=artist.id, platform="patreon",
|
||||||
url="https://patreon.com/alice",
|
url="https://patreon.com/alice",
|
||||||
)
|
)
|
||||||
|
# Simulate a prior walk's posts counter — start must clear it (plan #704).
|
||||||
|
await db.execute(
|
||||||
|
Source.__table__.update().where(Source.id == rec.id).values(
|
||||||
|
config_overrides={"_backfill_posts": 42},
|
||||||
|
)
|
||||||
|
)
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
updated = await svc.start_recovery(rec.id)
|
updated = await svc.start_recovery(rec.id)
|
||||||
assert updated.backfill_state == "running"
|
assert updated.backfill_state == "running"
|
||||||
assert updated.backfill_bypass_seen is True
|
assert updated.backfill_bypass_seen is True
|
||||||
|
assert updated.backfill_posts == 0 # cleared for a fresh walk
|
||||||
assert updated.backfill_runs_remaining == BACKFILL_MAX_CHUNKS
|
assert updated.backfill_runs_remaining == BACKFILL_MAX_CHUNKS
|
||||||
|
|
||||||
co = (await db.execute(
|
co = (await db.execute(
|
||||||
select(Source.config_overrides).where(Source.id == rec.id)
|
select(Source.config_overrides).where(Source.id == rec.id)
|
||||||
)).scalar_one()
|
)).scalar_one()
|
||||||
assert co.get("_backfill_bypass_seen") is True
|
assert co.get("_backfill_bypass_seen") is True
|
||||||
|
assert "_backfill_posts" not in co
|
||||||
|
|
||||||
# Stop clears the bypass flag too (shared lifecycle).
|
# Stop clears the bypass flag too (shared lifecycle).
|
||||||
stopped = await svc.stop_backfill(rec.id)
|
stopped = await svc.stop_backfill(rec.id)
|
||||||
|
|||||||
Reference in New Issue
Block a user