33ae5a6f7d
Splits pull+transcribe (now tower-side, eager) from extract+embed (stays on the dev scheduler). Three machine-coordination pieces land together because they reference each other: - v3 migration adds sources.claimed_by + claimed_at — observability + stale-claim recovery columns. The actual race-safety primitive is `SELECT ... FOR UPDATE SKIP LOCKED` in the new claim helper, so two machines can poll the queue without doubling work. - src/second_brain/claim.py owns the claim dance (claim_next_source, release_claim, reap_stale_claims). Both stage gates filter by source_type so the tower never grabs articles and the dev side never grabs videos. - src/second_brain/transcribe.py wraps faster-whisper (lazy-imported so it stays out of the dev install). resolve_settings() reads env > [whisper] block > defaults, falling back to int8 on cpu / float16 on cuda when compute_type is unspecified. Default model large-v3. - src/second_brain/scheduler/transcribe_worker.py is the long-running poll loop. Reads pipeline_settings every iteration so the dashboard's enable/window/max-items/max-video-length take effect within one cycle. Reaps stale claims at startup. SIGTERM-clean. DB-unreachable backs off with a log line; never crash-loops. - adapters/youtube.py drops the torch-whisper transcribe path; pull stays. Removes openai-whisper from the default deps and gates faster-whisper behind a new `tower` extra (uv sync --extra tower). - main.py: new `second-brain transcribe-worker` (--once for ad-hoc). `process` now article-only on the pull side but still picks up TRANSCRIBED of any source_type for the extract step. Live-verified: migration applies clean, transcribe-worker --once honors transcription_enabled=false gate.
51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
"""v3 second_brain: sources.claimed_by / claimed_at
|
|
|
|
Revision ID: c8f1d9e34b7a
|
|
Revises: 7b3e8a5c1f29
|
|
Create Date: 2026-05-25 12:30:00.000000
|
|
|
|
Adds a lightweight claim mechanism so the tower and dev sides never grab
|
|
the same row off the queue. The claim flow used by both workers is:
|
|
|
|
BEGIN;
|
|
SELECT id, status, source_type
|
|
FROM second_brain.sources
|
|
WHERE <filter>
|
|
AND claimed_by IS NULL
|
|
ORDER BY ingested_at ASC
|
|
LIMIT 1
|
|
FOR UPDATE SKIP LOCKED;
|
|
UPDATE second_brain.sources
|
|
SET claimed_by = :worker, claimed_at = now() AT TIME ZONE 'UTC'
|
|
WHERE id = :id;
|
|
COMMIT;
|
|
|
|
`FOR UPDATE SKIP LOCKED` is the actual race-safety primitive — the column
|
|
pair is observability (who has it, since when, for stale-claim recovery).
|
|
A partial index would help once we have a real queue depth; deliberately
|
|
not added now per Travis's "no gold-plating".
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
|
|
from alembic import op
|
|
|
|
revision: str = "c8f1d9e34b7a"
|
|
down_revision: str | Sequence[str] | None = "7b3e8a5c1f29"
|
|
branch_labels: str | Sequence[str] | None = None
|
|
depends_on: str | Sequence[str] | None = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
op.execute("SET search_path TO second_brain")
|
|
op.execute("ALTER TABLE sources ADD COLUMN claimed_by VARCHAR(100)")
|
|
op.execute("ALTER TABLE sources ADD COLUMN claimed_at TIMESTAMP")
|
|
|
|
|
|
def downgrade() -> None:
|
|
op.execute("SET search_path TO second_brain")
|
|
op.execute("ALTER TABLE sources DROP COLUMN IF EXISTS claimed_at")
|
|
op.execute("ALTER TABLE sources DROP COLUMN IF EXISTS claimed_by")
|