Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 42 additions & 18 deletions backend/app/post_content_queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -384,20 +384,13 @@ async def ensure_post_content_job(
POST_CONTENT_BACKFILL_CANDIDATE_SQL = f"""
select post.post_id, post.post_body
from source_post post
left join post_content_ingestion_job job on job.post_id = post.post_id
left join operations_case_analysis analysis
on analysis.post_id = post.post_id
and analysis.source_body_sha256 = job.source_body_sha256
left join post_product_analysis product_analysis
on product_analysis.post_id = post.post_id
and product_analysis.source_body_sha256 = job.source_body_sha256
left join (
select distinct project.post_id
from post_project_mention project
where nullif(btrim(project.ontology_iri), '') is not null
) ontology_project on ontology_project.post_id = post.post_id
where {SOURCE_POST_ELIGIBILITY_SQL.format(alias='post')}
and (job.post_id is null or job.status_code = $1)
and not exists (
select 1
from post_content_ingestion_job job
where job.post_id = post.post_id
and job.status_code is distinct from $1
)
and (
not exists (
select 1 from post_content_unit unit
Expand Down Expand Up @@ -436,14 +429,45 @@ async def ensure_post_content_job(
or structure.decision_source_code = 'unresolved'
)
))
or ($3::boolean and analysis.post_id is null)
or ($3::boolean and product_analysis.post_id is null)
or ($3::boolean and not exists (
select 1
from post_content_ingestion_job job
join operations_case_analysis analysis
on analysis.post_id = job.post_id
and analysis.source_body_sha256 = job.source_body_sha256
where job.post_id = post.post_id
))
or ($3::boolean and not exists (
select 1
from post_content_ingestion_job job
join post_product_analysis product_analysis
on product_analysis.post_id = job.post_id
and product_analysis.source_body_sha256 = job.source_body_sha256
where job.post_id = post.post_id
))
Comment thread
seonghobae marked this conversation as resolved.
)
and ($5::boolean = (
$3::boolean
and ontology_project.post_id is not null
and job.source_body_sha256 is not null
and analysis.post_id is null
and exists (
select 1
from post_project_mention project
where project.post_id = post.post_id
and nullif(btrim(project.ontology_iri), '') is not null
)
and exists (
select 1
from post_content_ingestion_job job
where job.post_id = post.post_id
and job.source_body_sha256 is not null
)
and not exists (
select 1
from post_content_ingestion_job job
join operations_case_analysis analysis
on analysis.post_id = job.post_id
and analysis.source_body_sha256 = job.source_body_sha256
where job.post_id = post.post_id
)
))
order by coalesce(post.event_occurred_at, post.created_at),
post.created_at,
Expand Down
8 changes: 8 additions & 0 deletions docs/adr/0098-valkey-backed-post-content-ingestion.md
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,14 @@ work or an unbounded HTTP request. Candidate selection and broker recovery are
independent: either failure is recorded and retried on the next cycle without
stopping the worker.

The bounded candidate scan uses the partial
`source_post_content_backfill_candidate_idx` on the candidate query's event-time
fallback and deterministic tie-breakers. Its partial predicate excludes drafts
and deleted rows; the query retains the shared source-context predicate. This
lets PostgreSQL stop after the requested ordered page instead of evaluating
content completeness across the whole source corpus. It does not change
eligibility or completeness semantics.

The CLI retains the same per-query bound. `--all-pages` repeats that governed
producer until the current candidate set is empty; progress remains visible in
the normalized job ledger after every page. Terminal failures are never reset
Expand Down
9 changes: 9 additions & 0 deletions migrations/0258_post_content_backfill_candidate_index.sql
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
-- Migration 0258 / ADR 0098: stop the bounded backfill scan in source order.
create index if not exists source_post_content_backfill_candidate_idx
on source_post (
coalesce(event_occurred_at, created_at),
created_at,
post_id
)
where nullif(btrim(source_draft_code), '') is null
and nullif(btrim(source_deleted_flag), '') is null;
Comment thread
seonghobae marked this conversation as resolved.
Comment thread
seonghobae marked this conversation as resolved.
28 changes: 28 additions & 0 deletions tests/test_post_content_backfill_schema.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
"""Static schema contract for the bounded post-content backfill scan."""

from pathlib import Path


MIGRATION = Path("migrations/0258_post_content_backfill_candidate_index.sql")


def test_backfill_candidate_index_matches_the_ordered_eligibility_scan() -> None:
"""The replay-safe partial index owns ordering and source eligibility."""
sql = " ".join(MIGRATION.read_text().lower().split())

assert "create index if not exists source_post_content_backfill_candidate_idx" in sql
assert (
"on source_post ( coalesce(event_occurred_at, created_at), created_at, post_id )"
in sql
)
for column in ("source_draft_code", "source_deleted_flag"):
assert f"nullif(btrim({column}), '')" in sql


def test_backfill_index_owns_its_stacked_migration_identity() -> None:
"""The index owns 0258 rather than reusing the parent stack's 0257."""

migrations = MIGRATION.parent

assert MIGRATION.name.startswith("0258_")
assert not (migrations / "0257_post_content_backfill_candidate_index.sql").exists()
8 changes: 4 additions & 4 deletions tests/test_post_content_queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,11 +65,11 @@ def transaction(self) -> Transaction:
async def fetch(self, query: str, *args: object) -> list[dict[str, str]]:
assert "source_draft_code" in query
assert "source_deleted_flag" in query
assert "job.post_id is null or job.status_code = $1" in query
assert "left join operations_case_analysis analysis" in query
assert "analysis.post_id = post.post_id" in query
assert "job.status_code is distinct from $1" in query
assert "join operations_case_analysis analysis" in query
assert "analysis.post_id = job.post_id" in query
assert "analysis.source_body_sha256 = job.source_body_sha256" in query
assert "left join post_product_analysis product_analysis" in query
assert "join post_product_analysis product_analysis" in query
assert "from post_project_mention project" in query
assert "nullif(btrim(project.ontology_iri), '') is not null" in query
assert "job.source_body_sha256 is not null" in query
Expand Down