From 90c6fed3836c00bbe0bb7a684af0644394fc3bd1 Mon Sep 17 00:00:00 2001 From: GF-2025 Date: Sun, 4 Oct 2026 16:59:34 +0800 Subject: [PATCH] fix(worker): a source deleted before processing must fail permanently, not starve the queue (#1364) * fix(worker): fail permanently when the source was deleted before processing Source.get() raises NotFoundError instead of returning None, so the 'if not source: raise ValueError' guard in process_source_command was unreachable. NotFoundError is not in the command's stop_on list, so the job was retried as transient (max_attempts=15, exponential_jitter, wait_max=120) and every job behind it starved until the worker process was restarted. Catch NotFoundError around that call and re-raise as ValueError, which is already this command's permanent-failure channel. Fixes #1363 * chore(worker): rebase on main, cover deleted transformations, changelog - Rebased onto main (import conflict with #1425). - A transformation deleted before the job runs hits the same NotFoundError path in process_source_command; it now fails permanently too (test added). - ruff format on the new test (CI gates formatting since #1429). - CHANGELOG entry. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy * fix(domain): only a missing record is NotFoundError in ObjectModel.get ObjectModel.get wrapped every exception from repo_query, including the RuntimeError it re-raises for retriable SurrealDB transaction conflicts, as NotFoundError. With process_source now treating NotFoundError as permanent, a transient conflict would have failed the job for good. get() now lets NotFoundError / InvalidInputError through and raises DatabaseOperationError (chained) for anything else. The permanent-failure messages are static; the underlying error stays as the exception's cause for the worker log. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy * fix(worker): run_transformation fails permanently on a deleted source too NotFoundError (now only raised for a genuinely missing record) joins run_transformation's stop_on and permanent-failure arm. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy * docs(changelog): rebase entry onto the consolidated Unreleased section Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy * fix(worker): a source deleted mid-processing also fails permanently save_source re-reads the source after extraction; a deletion in that window raised NotFoundError from inside the graph, which process_source retried as transient. NotFoundError (now only a genuinely missing record) joins its stop_on and gets a permanent-failure arm. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy * docs(adr): ADR-013 for the ObjectModel.get error contract Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy * fix(embeddings): a record deleted before embedding is a permanent failure Completes the review on #1364: embed_* commands treat NotFoundError (now only a genuinely missing record) like their other permanent failures instead of retrying it; a DatabaseOperationError still retries. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy * docs(embeddings): NotFoundError is a permanent failure too Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GBYpxzSS62Y3TZkgB9NUZy --------- Co-authored-by: Luis Novo Co-authored-by: Claude Opus 5.5 --- CHANGELOG.md | 1 + commands/embedding_commands.py | 31 ++- commands/source_commands.py | 38 +++- .../ADR-013-objectmodel-get-error-contract.md | 30 +++ docs/7-DEVELOPMENT/decisions/README.md | 1 + open_notebook/domain/base.py | 7 +- .../test_source_deleted_before_processing.py | 204 ++++++++++++++++++ 7 files changed, 296 insertions(+), 16 deletions(-) create mode 100644 docs/7-DEVELOPMENT/decisions/ADR-013-objectmodel-get-error-contract.md create mode 100644 tests/test_source_deleted_before_processing.py diff --git a/CHANGELOG.md b/CHANGELOG.md index da957da6..3c895b43 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Credential encryption now uses PBKDF2-HMAC-SHA256 (600k iterations) with versioned `pbkdf2v1:` ciphertext.** New keys are written in the new format; legacy values keep decrypting, and a one-shot `POST /api/credentials/migrate-encryption` pass upgrades stored keys (idempotent, fail-closed per record). Downgrade warning: versions predating this change cannot read the new format — back up the database before migrating (#1317) ### Fixed +- **The worker no longer stalls when a source is deleted before it is processed.** `Source.get()` raises `NotFoundError` for a missing record, so the job was retried as a transient failure (up to 15 attempts with backoff) and every job behind it waited; a missing source or transformation now fails the job immediately. Relatedly, loading a record no longer reports a database failure (e.g. a SurrealDB transaction conflict) as "not found": it raises a database error, so it stays retryable in jobs and returns 500 instead of 404 in the API (#1363) - **Model calls default to a 180-second timeout.** esperanto 2.28 now enforces `ESPERANTO_LLM_TIMEOUT` on every provider (60 s by default); before, most providers used their SDK default and Ollama waited indefinitely. Open Notebook now sets 180 s when the variable is unset, so long answers and slower models aren't cut off at 60 s. **Ollama and other slow local setups** that relied on unlimited waits should set `ESPERANTO_LLM_TIMEOUT` (and `API_CLIENT_TIMEOUT` above it) explicitly (#1434) - Batch large embedding inserts into groups of 50 records and reuse one database connection across batches to reduce oversized WebSocket payloads and repeated sign-ins. - `GET /api/notebooks` now normalizes `order_by` through the shared `_validate_order_by()` guard, keeping its stricter `name`/`created`/`updated` allowlist on top, so the two ORDER BY validators can no longer drift (#1416) diff --git a/commands/embedding_commands.py b/commands/embedding_commands.py index 14fae108..98e99920 100644 --- a/commands/embedding_commands.py +++ b/commands/embedding_commands.py @@ -16,13 +16,17 @@ from surreal_commands import CommandInput, CommandOutput, command, submit_comman from open_notebook.ai.models import model_manager from open_notebook.database.repository import ensure_record_id, repo_insert, repo_query from open_notebook.domain.notebook import Note, Source, SourceInsight -from open_notebook.exceptions import ConfigurationError, ContextLengthExceededError +from open_notebook.exceptions import ( + ConfigurationError, + ContextLengthExceededError, + NotFoundError, +) from open_notebook.utils.chunking import ContentType, chunk_text, detect_content_type from open_notebook.utils.embedding import generate_embedding, generate_embeddings # NOTE: `stop_on` below can never trigger in practice — each command catches -# ValueError internally and returns success=False instead of raising, so the -# retry layer never sees it. Kept as-is on purpose; to be revisited in a +# ValueError (and NotFoundError) internally and returns success=False instead +# of raising, so the retry layer never sees it. Kept as-is on purpose; to be revisited in a # dedicated error-handling PR. EMBED_RETRY_CONFIG = { "max_attempts": 5, @@ -66,7 +70,8 @@ async def _embed_record( Returns: (extra_output_fields, processing_time, error_message) extra_output_fields is None and error_message is set on permanent - (ValueError) failure. Transient failures re-raise so the retry layer + (ValueError, or NotFoundError for a record deleted before the job ran) + failure. Transient failures re-raise so the retry layer can handle them. """ start_time = time.time() @@ -82,8 +87,10 @@ async def _embed_record( ) return extra_fields, processing_time, None - except ValueError as e: - # Permanent failure - don't retry + except (ValueError, NotFoundError) as e: + # Permanent failure - don't retry. NotFoundError means the record was + # deleted before the job ran (ObjectModel.get raises it only for a + # missing record; DB failures are DatabaseOperationError and retry). processing_time = time.time() - start_time cmd_id = get_command_id(input_data) logger.error(f"Failed to embed {kind} {record_id} (command: {cmd_id}): {e}") @@ -233,7 +240,8 @@ async def embed_note_command(input_data: EmbedNoteInput) -> EmbedNoteOutput: Retry Strategy: - Retries up to 5 times for transient failures (network, timeout, etc.) - Uses exponential-jitter backoff (1-60s) - - Does NOT retry permanent failures (ValueError for validation errors) + - Does NOT retry permanent failures (ValueError for validation errors, + NotFoundError for a record deleted before the job ran) """ async def embed() -> Tuple[Dict[str, Any], str]: @@ -275,7 +283,8 @@ async def embed_insight_command(input_data: EmbedInsightInput) -> EmbedInsightOu Retry Strategy: - Retries up to 5 times for transient failures (network, timeout, etc.) - Uses exponential-jitter backoff (1-60s) - - Does NOT retry permanent failures (ValueError for validation errors) + - Does NOT retry permanent failures (ValueError for validation errors, + NotFoundError for a record deleted before the job ran) """ async def embed() -> Tuple[Dict[str, Any], str]: @@ -320,7 +329,8 @@ async def embed_source_command(input_data: EmbedSourceInput) -> EmbedSourceOutpu Retry Strategy: - Retries up to 5 times for transient failures (network, timeout, etc.) - Uses exponential-jitter backoff (1-60s) - - Does NOT retry permanent failures (ValueError for validation errors) + - Does NOT retry permanent failures (ValueError for validation errors, + NotFoundError for a record deleted before the job ran) """ async def embed() -> Tuple[Dict[str, Any], str]: @@ -440,7 +450,8 @@ async def create_insight_command( Retry Strategy: - Retries up to 5 times for transient failures (network, timeout, etc.) - Uses exponential-jitter backoff (1-60s) - - Does NOT retry permanent failures (ValueError for validation errors) + - Does NOT retry permanent failures (ValueError for validation errors, + NotFoundError for a record deleted before the job ran) """ start_time = time.time() diff --git a/commands/source_commands.py b/commands/source_commands.py index 8f59929e..c3e999f4 100644 --- a/commands/source_commands.py +++ b/commands/source_commands.py @@ -13,6 +13,7 @@ from open_notebook.exceptions import ( ContextLengthExceededError, IncompleteGenerationError, InvalidInputError, + NotFoundError, ) try: @@ -53,7 +54,10 @@ class SourceProcessingOutput(CommandOutput): ConfigurationError, ContextLengthExceededError, IncompleteGenerationError, - ], # Don't retry validation/config errors or incomplete generations + NotFoundError, + ], # Don't retry validation/config errors, incomplete generations, or a + # record deleted mid-processing (NotFoundError means "missing", never a + # DB failure: ObjectModel.get raises DatabaseOperationError for those) "retry_log_level": "debug", # Avoid log noise during transaction conflicts }, ) @@ -75,7 +79,11 @@ async def process_source_command( transformations = [] for trans_id in input_data.transformations: logger.info(f"Loading transformation: {trans_id}") - transformation = await Transformation.get(trans_id) + try: + transformation = await Transformation.get(trans_id) + except NotFoundError as e: + # Same as a deleted source below: permanent, not transient. + raise ValueError(f"Transformation '{trans_id}' no longer exists") from e if not transformation: raise ValueError(f"Transformation '{trans_id}' not found") transformations.append(transformation) @@ -83,7 +91,18 @@ async def process_source_command( logger.info(f"Loaded {len(transformations)} transformations") # 2. Get existing source record to update its command field - source = await Source.get(input_data.source_id) + try: + source = await Source.get(input_data.source_id) + except NotFoundError as e: + # The source was removed after this job was queued (e.g. a sync tool + # deleted it, or the record was cleaned up). The job can never + # succeed: raise a permanent error (ValueError is in `stop_on`) so + # surreal-commands marks it failed instead of spending 15 retries + # with exponential backoff, which starves every job behind it. + raise ValueError( + f"Source '{input_data.source_id}' no longer exists " + "(deleted before processing?)" + ) from e if not source: raise ValueError(f"Source '{input_data.source_id}' not found") @@ -142,6 +161,13 @@ async def process_source_command( f"Generation failed (permanent) for source {input_data.source_id}: {e}" ) raise # Preserve failed job status; stop_on prevents automatic retries. + except NotFoundError as e: + # E.g. the source was deleted while extraction ran (save_source re-reads + # it). Permanent: stop_on prevents retries that would starve the queue. + logger.error( + f"Source processing failed (permanent), record no longer exists: {e}" + ) + raise except ValueError as e: # Validation errors are permanent failures. Re-raise so surreal-commands # marks the job as `failed` (stop_on=[ValueError] already prevents @@ -193,7 +219,9 @@ class RunTransformationOutput(CommandOutput): ContextLengthExceededError, IncompleteGenerationError, InvalidInputError, - ], # Don't retry validation/config errors or incomplete generations + NotFoundError, + ], # Don't retry validation/config errors, incomplete generations, or + # a source/transformation deleted before the job ran "retry_log_level": "warning", }, ) @@ -257,7 +285,7 @@ async def run_transformation_command( processing_time=processing_time, ) - except (IncompleteGenerationError, InvalidInputError) as e: + except (IncompleteGenerationError, InvalidInputError, NotFoundError) as e: # e.g. the source has no text to transform logger.error( f"Generation failed (permanent) for transformation " diff --git a/docs/7-DEVELOPMENT/decisions/ADR-013-objectmodel-get-error-contract.md b/docs/7-DEVELOPMENT/decisions/ADR-013-objectmodel-get-error-contract.md new file mode 100644 index 00000000..27a4569e --- /dev/null +++ b/docs/7-DEVELOPMENT/decisions/ADR-013-objectmodel-get-error-contract.md @@ -0,0 +1,30 @@ +# ADR-013: ObjectModel.get raises NotFoundError only for a missing record + +- **Status**: Accepted +- **Date**: 2026-10 +- **Related**: #1363, PR #1364 + +## Context + +`ObjectModel.get()` is the shared fetch boundary for every domain model. It wrapped **every** exception from `repo_query` as `NotFoundError("Object with id ... not found - {e}")`, including connection errors and the `RuntimeError` that `repo_query` re-raises for retriable SurrealDB v2 transaction conflicts. Callers treat `NotFoundError` as permanent: routers map it to 404, and since #1363 the background commands treat it as a non-retryable job failure. Under the old wrapping, a transient database hiccup would have been reported as "not found" and never retried. + +## Decision + +**`ObjectModel.get()` raises `NotFoundError` only when the record does not exist.** `InvalidInputError` (empty id, unknown table) passes through unchanged. Any other failure is raised as `DatabaseOperationError`, chained to the original exception. + +Consequences for callers: + +- **Commands:** `NotFoundError` means "the record was deleted" and is permanent. `process_source` and `run_transformation` have it in `stop_on`. `DatabaseOperationError` stays retryable. +- **API:** a missing record is still 404. A database failure during a fetch becomes a 500 (`OpenNotebookError` handler) instead of a misleading 404. + +## Alternatives considered + +- **Keep the wrapping and catch `NotFoundError` per command**: rejected, because it makes every transient conflict permanent, exactly the failure the worker's retries exist for. +- **Match the error message to tell the two apart**: rejected as fragile string coupling. +- **Return `None` for a missing record**: rejected. It's a larger API change, and the `if not x` guards would have to come back everywhere. + +## Consequences + +- A command can safely treat `NotFoundError` as permanent (deleted before or during processing) without starving the queue on DB conflicts. +- API clients see 500, not 404, for database outages on single-record fetches. That's correct, but it's visible if a client relied on the old 404. +- New fetch helpers should follow the same split: "missing" vs "failed". diff --git a/docs/7-DEVELOPMENT/decisions/README.md b/docs/7-DEVELOPMENT/decisions/README.md index 4f37541d..789c0084 100644 --- a/docs/7-DEVELOPMENT/decisions/README.md +++ b/docs/7-DEVELOPMENT/decisions/README.md @@ -53,5 +53,6 @@ What this makes easier, what it makes harder, what to watch. (bullets) | [ADR-009](ADR-009-pbkdf2-credential-encryption.md) | PBKDF2 credential key derivation with versioned ciphertext | Accepted | | [ADR-011](ADR-011-design-token-system.md) | Visual identity is a token contract in globals.css, reviewed through /dev/design | Accepted | | [ADR-012](ADR-012-provider-endpoint-overrides.md) | Provider endpoint overrides are declared in the registry | Accepted | +| [ADR-013](ADR-013-objectmodel-get-error-contract.md) | ObjectModel.get raises NotFoundError only for a missing record | Accepted | | [PDR-001](PDR-001-single-user-first.md) | Single-user first; don't preclude multi-user | Accepted | | [PDR-002](PDR-002-provider-agnostic-core.md) | Provider-agnostic core by default | Accepted | diff --git a/open_notebook/domain/base.py b/open_notebook/domain/base.py index a94cfa03..ad6e2e58 100644 --- a/open_notebook/domain/base.py +++ b/open_notebook/domain/base.py @@ -125,10 +125,15 @@ class ObjectModel(BaseModel): return target_class(**result[0]) else: raise NotFoundError(f"{table_name} with id {id} not found") + except (NotFoundError, InvalidInputError): + raise except Exception as e: + # A database failure (connection, SurrealDB transaction conflict...) + # is not "not found": callers treat NotFoundError as permanent (404, + # no retry), so surface it as DatabaseOperationError instead. logger.error(f"Error fetching object with id {id}: {str(e)}") logger.exception(e) - raise NotFoundError(f"Object with id {id} not found - {str(e)}") + raise DatabaseOperationError(f"Failed to fetch object with id {id}") from e @classmethod def _get_class_by_table_name(cls, table_name: str) -> Optional[Type["ObjectModel"]]: diff --git a/tests/test_source_deleted_before_processing.py b/tests/test_source_deleted_before_processing.py new file mode 100644 index 00000000..dd689544 --- /dev/null +++ b/tests/test_source_deleted_before_processing.py @@ -0,0 +1,204 @@ +"""A source deleted before its process_source job runs must fail permanently. + +Regression for the worker-starvation bug: `Source.get()` raises `NotFoundError` +instead of returning `None`, so the `if not source: raise ValueError(...)` guard in +`process_source_command` never fired. `NotFoundError` is not in the command's +`stop_on` list, so the job was retried as a transient failure (up to 15 attempts +with exponential backoff, waits up to 120s) and every job behind it starved — +the worker stopped consuming jobs until the process was restarted. +""" + +from unittest.mock import AsyncMock, patch + +import pytest + +import commands # noqa: F401 -- import registers the @command decorators +from open_notebook.exceptions import NotFoundError + + +@pytest.mark.asyncio +async def test_missing_source_raises_permanent_error(): + """NotFoundError from Source.get must surface as ValueError (terminal).""" + from commands.source_commands import SourceProcessingInput, process_source_command + + input_data = SourceProcessingInput( + source_id="source:does-not-exist", + content_state={"file_path": "/tmp/whatever.md"}, + notebook_ids=["notebook:whatever"], + transformations=[], + embed=True, + ) + + with patch( + "commands.source_commands.Source.get", + new=AsyncMock( + side_effect=NotFoundError("source with id source:does-not-exist not found") + ), + ): + with pytest.raises(ValueError) as excinfo: + await process_source_command(input_data) + + assert "no longer exists" in str(excinfo.value) + # The retry policy keys off the exception type: ValueError is in `stop_on`, + # NotFoundError is not — that difference is the whole fix. + assert not isinstance(excinfo.value, NotFoundError) + + +@pytest.mark.asyncio +async def test_missing_transformation_raises_permanent_error(): + """A transformation deleted before the job runs is just as permanent.""" + from commands.source_commands import SourceProcessingInput, process_source_command + + input_data = SourceProcessingInput( + source_id="source:abc", + content_state={"file_path": "/tmp/whatever.md"}, + notebook_ids=["notebook:whatever"], + transformations=["transformation:gone"], + embed=True, + ) + + with patch( + "commands.source_commands.Transformation.get", + new=AsyncMock(side_effect=NotFoundError("transformation not found")), + ): + with pytest.raises(ValueError, match="no longer exists"): + await process_source_command(input_data) + + +@pytest.mark.asyncio +async def test_object_get_reports_db_failures_as_database_errors(): + """ObjectModel.get used to wrap every exception (e.g. a retriable SurrealDB + transaction conflict) as NotFoundError, so the fix above would have made + transient failures permanent. Only a missing record is NotFoundError.""" + from open_notebook.domain.notebook import Source + from open_notebook.exceptions import DatabaseOperationError + + with patch( + "open_notebook.domain.base.repo_query", + new=AsyncMock(side_effect=RuntimeError("Transaction conflict: retry")), + ): + with pytest.raises(DatabaseOperationError): + await Source.get("source:abc") + + with patch("open_notebook.domain.base.repo_query", new=AsyncMock(return_value=[])): + with pytest.raises(NotFoundError): + await Source.get("source:abc") + + +@pytest.mark.asyncio +async def test_transient_db_failure_stays_retryable(): + """A database failure while loading the source is re-raised as is (not + ValueError), so the worker's retry policy still applies.""" + from commands.source_commands import SourceProcessingInput, process_source_command + from open_notebook.exceptions import DatabaseOperationError + + input_data = SourceProcessingInput( + source_id="source:abc", + content_state={"file_path": "/tmp/whatever.md"}, + notebook_ids=["notebook:whatever"], + transformations=[], + embed=True, + ) + + with patch( + "commands.source_commands.Source.get", + new=AsyncMock(side_effect=DatabaseOperationError("Failed to fetch")), + ): + with pytest.raises(DatabaseOperationError): + await process_source_command(input_data) + + +@pytest.mark.asyncio +async def test_run_transformation_fails_permanently_when_source_is_gone(): + """run_transformation had the same deletion-before-processing gap.""" + from surreal_commands import registry + + from commands.source_commands import ( + RunTransformationInput, + run_transformation_command, + ) + + with patch( + "commands.source_commands.Source.get", + new=AsyncMock(side_effect=NotFoundError("source not found")), + ): + with pytest.raises(NotFoundError): + await run_transformation_command( + RunTransformationInput( + source_id="source:gone", transformation_id="transformation:t1" + ) + ) + + command = registry.get_command("open_notebook", "run_transformation") + assert command is not None + assert NotFoundError in command.retry_config.stop_on + + +@pytest.mark.asyncio +async def test_source_deleted_during_processing_fails_permanently(): + """save_source re-reads the source after extraction; a deletion in that + window raises NotFoundError from inside the graph, which must not be + retried either.""" + from types import SimpleNamespace + + from surreal_commands import registry + + from commands.source_commands import SourceProcessingInput, process_source_command + + source = SimpleNamespace(id="source:abc", command=None, save=AsyncMock()) + with ( + patch( + "commands.source_commands.Source.get", new=AsyncMock(return_value=source) + ), + patch( + "commands.source_commands.source_graph.ainvoke", + new=AsyncMock( + side_effect=NotFoundError("source with id source:abc not found") + ), + ), + ): + with pytest.raises(NotFoundError): + await process_source_command( + SourceProcessingInput( + source_id="source:abc", + content_state={"file_path": "/tmp/x.md"}, + notebook_ids=[], + transformations=[], + embed=False, + ) + ) + + command = registry.get_command("open_notebook", "process_source") + assert command is not None + assert NotFoundError in command.retry_config.stop_on + + +@pytest.mark.asyncio +async def test_embedding_a_deleted_source_fails_permanently(): + """embed_source had the same gap (lfnovo's review on #1364): a record + deleted before the job ran is a permanent failure, not a retry.""" + from commands.embedding_commands import EmbedSourceInput, embed_source_command + + with patch( + "commands.embedding_commands.Source.get", + new=AsyncMock( + side_effect=NotFoundError("source with id source:gone not found") + ), + ): + result = await embed_source_command(EmbedSourceInput(source_id="source:gone")) + + assert result.success is False + assert "not found" in (result.error_message or "") + + +@pytest.mark.asyncio +async def test_embedding_db_failure_stays_retryable(): + from commands.embedding_commands import EmbedSourceInput, embed_source_command + from open_notebook.exceptions import DatabaseOperationError + + with patch( + "commands.embedding_commands.Source.get", + new=AsyncMock(side_effect=DatabaseOperationError("Failed to fetch")), + ): + with pytest.raises(DatabaseOperationError): + await embed_source_command(EmbedSourceInput(source_id="source:abc"))