From 9d8ce36075b643adddb2aa12387fcba631106e2a Mon Sep 17 00:00:00 2001 From: DevBehindYou Date: Thu, 29 Oct 2026 21:57:51 +0530 Subject: [PATCH] Share conversion time budgets across dispatch and workers --- backendN/app/common/app_factory.py | 11 ++ backendN/app/common/deadlines.py | 91 +++++++++++++ backendN/app/common/security.py | 2 + backendN/app/common/storage.py | 9 +- backendN/app/pdf_tasks.py | 12 +- backendN/tests/test_deadlines.py | 136 ++++++++++++++++++++ backendN/tests/test_pdf_tasks.py | 16 +++ backendO/app/common/app_factory.py | 11 ++ backendO/app/common/deadlines.py | 91 +++++++++++++ backendO/app/common/security.py | 2 + backendO/app/common/storage.py | 9 +- backendO/app/converter.py | 12 +- backendO/tests/test_deadlines.py | 47 +++++++ backendZ/app/archive.py | 5 + backendZ/app/archive_tasks.py | 13 +- backendZ/app/common/app_factory.py | 11 ++ backendZ/app/common/deadlines.py | 91 +++++++++++++ backendZ/app/common/security.py | 2 + backendZ/app/common/storage.py | 9 +- backendZ/tests/test_deadlines.py | 31 +++++ docs/LLMISER_OPTIMIZATION_PLAN.md | 2 + docs/LLMISER_SHARED_DEADLINES.md | 29 +++++ docs/START_HERE.md | 2 + docs/ai-handover/HANDOVER.md | 2 + docs/ai-handover/PROGRESS.md | 2 + frontend/app/api/convert/route.js | 3 + frontend/app/api/jobs/[id]/advance/route.js | 6 +- frontend/app/api/jobs/[id]/start/route.js | 6 +- frontend/app/api/jobs/tick/route.js | 7 +- frontend/lib/server/dispatcher.js | 30 +++-- frontend/lib/server/jobQueue.js | 9 +- frontend/lib/server/jobService.js | 9 +- frontend/lib/server/supabaseRest.js | 6 +- frontend/lib/server/timeBudget.js | 23 ++++ frontend/test/timeBudget.test.mjs | 92 +++++++++++++ 35 files changed, 806 insertions(+), 33 deletions(-) create mode 100644 backendN/app/common/deadlines.py create mode 100644 backendN/tests/test_deadlines.py create mode 100644 backendO/app/common/deadlines.py create mode 100644 backendO/tests/test_deadlines.py create mode 100644 backendZ/app/common/deadlines.py create mode 100644 backendZ/tests/test_deadlines.py create mode 100644 docs/LLMISER_SHARED_DEADLINES.md create mode 100644 frontend/lib/server/timeBudget.js create mode 100644 frontend/test/timeBudget.test.mjs diff --git a/backendN/app/common/app_factory.py b/backendN/app/common/app_factory.py index 08f92f8..a0c5bda 100644 --- a/backendN/app/common/app_factory.py +++ b/backendN/app/common/app_factory.py @@ -26,6 +26,7 @@ from .config import Settings, deployment_revision from .errors import ConversionRejected, EngineUnavailable +from .deadlines import DeadlineExceeded, DeadlineMiddleware, check_budget, compute_checkpoint from .files import read_limited, sanitize_filename, split_name, title_from_stem from .profiles import PROFILES, count_words, finalize_markdown from .security import require_internal_secret @@ -75,6 +76,7 @@ def create_app(settings: Settings, converter: Converter, storage: StorageClient # No interactive docs or schema: these instances are internal compute, # not a public API surface. app = FastAPI(title=f"LLMiser backend ({settings.role})", docs_url=None, redoc_url=None, openapi_url=None) + app.add_middleware(DeadlineMiddleware) app.state.settings = settings app.state.converter = converter app.state.storage = storage if storage is not None else StorageClient.from_env() @@ -95,9 +97,12 @@ def run(original_name: str, profile: str, job_id: str, load_bytes, *, raw: bool if profile not in PROFILES: profile = "Standard" data = load_bytes() + compute_checkpoint() converter.validate(data, ext) title = title_from_stem(stem) + compute_checkpoint() markdown, warning = converter.convert(data, ext, filename=original_name, title=title) + compute_checkpoint() if raw: content = markdown.strip() + "\n" words = count_words(content) @@ -112,6 +117,9 @@ def run(original_name: str, profile: str, job_id: str, load_bytes, *, raw: bool profile=profile, ) outcome = "ok" + except DeadlineExceeded: + outcome = "deadline" + raise except ConversionRejected as err: outcome = err.code raise HTTPException(status_code=err.status_code, detail=err.detail) from None @@ -214,9 +222,11 @@ def process(req: ProcessRequest) -> dict: ) write_started = time.perf_counter() + check_budget() body = content.encode("utf-8") try: storage.upload(req.output_path, body, "text/markdown; charset=utf-8") + check_budget() except StorageError as err: # The conversion worked but the result isn't stored: never report # success. Transient → the dispatcher may retry on the peer. @@ -248,6 +258,7 @@ def convert( job_id, lambda: read_limited(file.file, settings.max_upload_bytes), ) + check_budget() return {**fields, "content": content} return app diff --git a/backendN/app/common/deadlines.py b/backendN/app/common/deadlines.py new file mode 100644 index 0000000..116ae31 --- /dev/null +++ b/backendN/app/common/deadlines.py @@ -0,0 +1,91 @@ +"""Optional authenticated per-request budget; legacy callers keep their limits.""" +from contextvars import ContextVar +from dataclasses import dataclass +from functools import wraps +import re +import time + +from fastapi import HTTPException + +HEADER = b"x-llmiser-deadline-ms" +FINALIZE_RESERVE_S = 10.0 +MAX_BUDGET_S = 300.0 +_current = ContextVar("processing_deadline", default=None) + + +class DeadlineExceeded(HTTPException): + def __init__(self): + super().__init__(504, "Conversion timed out. Try a smaller file or retry later.", + headers={"X-LLMiser-Deadline-Exceeded": "1"}) + + +@dataclass(frozen=True) +class Budget: + end: float + + def remaining(self): + return max(0.0, self.end - time.monotonic()) + + +class DeadlineMiddleware: + """Set context before body parsing/thread admission; auth validates it later.""" + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + value = None + if scope["type"] == "http" and scope.get("path", "").startswith("/api/v1/internal/"): + headers = [v for k, v in scope.get("headers", []) if k.lower() == HEADER] + if headers: + raw = headers[0] + if len(headers) != 1 or not re.fullmatch(rb"[0-9]{1,16}", raw): + value = "invalid" + else: + seconds = min(MAX_BUDGET_S, max(0.0, int(raw) / 1000 - time.time())) + value = Budget(time.monotonic() + seconds) + token = _current.set(value) + try: + await self.app(scope, receive, send) + finally: + _current.reset(token) + + +def current_budget(): + value = _current.get() + if value == "invalid": + raise HTTPException(400, "Invalid processing deadline") + return value + + +def check_budget(reserve=0.0): + budget = current_budget() + if budget is not None and budget.remaining() <= reserve: + raise DeadlineExceeded() + + +def compute_checkpoint(): + check_budget(FINALIZE_RESERVE_S) + + +def timeout_seconds(default, *, reserve=0.0): + budget = current_budget() + if budget is None: + return default + left = budget.remaining() - reserve + if left <= 0: + raise DeadlineExceeded() + return min(default, left) + + +def budgeted_worker(fn): + """ThreadPoolExecutor does not inherit the request's ContextVar.""" + value = _current.get() + @wraps(fn) + def run(*args, **kwargs): + token = _current.set(value) + try: + check_budget() + return fn(*args, **kwargs) + finally: + _current.reset(token) + return run diff --git a/backendN/app/common/security.py b/backendN/app/common/security.py index f1213db..bc36944 100644 --- a/backendN/app/common/security.py +++ b/backendN/app/common/security.py @@ -5,6 +5,7 @@ import hmac from fastapi import Header, HTTPException, Request +from .deadlines import check_budget def require_internal_secret( @@ -23,3 +24,4 @@ def require_internal_secret( x_internal_secret.encode(), expected.encode() ): raise HTTPException(status_code=401, detail="Unauthorized") + check_budget() diff --git a/backendN/app/common/storage.py b/backendN/app/common/storage.py index da6b48c..584e4d4 100644 --- a/backendN/app/common/storage.py +++ b/backendN/app/common/storage.py @@ -16,6 +16,7 @@ from urllib.parse import quote import httpx +from .deadlines import check_budget, timeout_seconds # Backends only touch the fixed object layout of a job (docs/BACKGROUND_JOBS.md). # Anything else is refused, so a bug or a compromised caller can't read or @@ -54,6 +55,7 @@ class StorageClient: def __init__(self, base_url: str, service_key: str, bucket: str, *, timeout: float = 30.0, transport=None): self._base = base_url.rstrip("/") + "/storage/v1" self._bucket = bucket + self._timeout_s = timeout self._http = httpx.Client( timeout=timeout, transport=transport, @@ -73,19 +75,21 @@ def _url(self, path: str) -> str: def download(self, path: str, max_bytes: int) -> bytes: try: - with self._http.stream("GET", self._url(path)) as res: + with self._http.stream("GET", self._url(path), timeout=timeout_seconds(self._timeout_s)) as res: if res.status_code in (400, 404): raise StorageError("Input object not found", transient=False, status=res.status_code) if res.status_code >= 400: raise StorageError(f"Storage read failed ({res.status_code})", transient=res.status_code >= 500, status=res.status_code) chunks, size = [], 0 for chunk in res.iter_bytes(): + check_budget() size += len(chunk) if size > max_bytes: raise StorageError("Input object exceeds the size limit", transient=False, status=413) chunks.append(chunk) return b"".join(chunks) except httpx.HTTPError as err: + check_budget() raise StorageError(f"Storage unreachable: {type(err).__name__}", transient=True) from None def upload(self, path: str, data: bytes, content_type: str) -> None: @@ -94,9 +98,12 @@ def upload(self, path: str, data: bytes, content_type: str) -> None: res = self._http.post( self._url(path), content=data, + timeout=timeout_seconds(self._timeout_s), headers={"Content-Type": content_type, "x-upsert": "true", "Cache-Control": "no-store"}, ) except httpx.HTTPError as err: + check_budget() raise StorageError(f"Storage unreachable: {type(err).__name__}", transient=True) from None if res.status_code >= 400: raise StorageError(f"Storage write failed ({res.status_code})", transient=res.status_code >= 500, status=res.status_code) + check_budget() diff --git a/backendN/app/pdf_tasks.py b/backendN/app/pdf_tasks.py index 33c15df..badb284 100644 --- a/backendN/app/pdf_tasks.py +++ b/backendN/app/pdf_tasks.py @@ -34,6 +34,7 @@ from app.common.files import sanitize_filename, split_name, title_from_stem from app.common.profiles import PROFILES, finalize_markdown from app.common.security import require_internal_secret +from app.common.deadlines import budgeted_worker, check_budget, compute_checkpoint from app.common.storage import StorageError logger = logging.getLogger("mdify.pdf") @@ -64,6 +65,7 @@ def _pdfium_serialized(fn): @wraps(fn) def run(*args, **kwargs): with _PDFIUM_LOCK: + compute_checkpoint() return fn(*args, **kwargs) return run @@ -89,6 +91,7 @@ def classify_pages(pdf) -> list[PageInfo]: pages = [] for index in range(len(pdf)): + compute_checkpoint() page = pdf[index] try: width, height = page.get_size() @@ -206,6 +209,7 @@ def analyze(req: ProcessRequest) -> dict: try: converter.validate(data, "pdf") with _PDFIUM_LOCK: + compute_checkpoint() pdf = pdfium.PdfDocument(data) except ConversionRejected as err: raise HTTPException(status_code=err.status_code, detail=err.detail) from None @@ -235,6 +239,7 @@ def analyze(req: ProcessRequest) -> dict: skipped = scans[MAX_OCR_PAGES:] segments, scan_units = [], [] for kind, first, last in page_runs(pages): + compute_checkpoint() node_id = str(uuid.uuid4()) if kind == "native" or first not in ocr_pages: if kind == "scan": @@ -242,6 +247,7 @@ def analyze(req: ProcessRequest) -> dict: markdown, _warning = converter.convert( sub_pdf(pdf, first, last), "pdf", filename=name, title=name, note_low_text=False ) + compute_checkpoint() path = f"jobs/{job_id}/nodes/{node_id}/result.md" store.upload(path, (markdown.strip() + "\n").encode("utf-8"), "text/markdown; charset=utf-8") segments.append({"node_id": node_id, "page_from": first, "page_to": last, "output_path": path}) @@ -302,6 +308,7 @@ def merge(req: MergeRequest) -> dict: profile = req.profile if req.profile in PROFILES else "Standard" def load(part: MergePart) -> tuple[MergePart, str | None]: + compute_checkpoint() try: return part, store.download(part.path, settings.max_upload_bytes).decode("utf-8", "replace") except StorageError as err: @@ -311,12 +318,13 @@ def load(part: MergePart) -> tuple[MergePart, str | None]: try: with ThreadPoolExecutor(max_workers=UPLOAD_WORKERS) as pool: - loaded = list(pool.map(load, sorted(req.parts, key=lambda p: p.page_from))) + loaded = list(pool.map(budgeted_worker(load), sorted(req.parts, key=lambda p: p.page_from))) except StorageError: raise HTTPException(status_code=503, detail="Storage temporarily unavailable") from None blocks, failed_pages = [], [] for part, text in loaded: + compute_checkpoint() if part.kind == "ocr": label = f"" if text is None or not text.strip(): @@ -333,6 +341,7 @@ def load(part: MergePart) -> tuple[MergePart, str | None]: f"(LLMiser reads up to {MAX_OCR_PAGES} scanned pages per PDF): {listed}." ) + compute_checkpoint() content, words, tokens = finalize_markdown( "\n\n".join(blocks) or "_No text found in this document._", title=title_from_stem(stem), @@ -342,6 +351,7 @@ def load(part: MergePart) -> tuple[MergePart, str | None]: profile=profile, ) body = content.encode("utf-8") + check_budget() try: store.upload(req.output_path, body, "text/markdown; charset=utf-8") except StorageError as err: diff --git a/backendN/tests/test_deadlines.py b/backendN/tests/test_deadlines.py new file mode 100644 index 0000000..9c51eed --- /dev/null +++ b/backendN/tests/test_deadlines.py @@ -0,0 +1,136 @@ +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +import time +import threading +from types import SimpleNamespace +import uuid + +from fastapi.testclient import TestClient +import httpx +import pytest + +from app.common import deadlines as budget +from app.common.app_factory import create_app +from app.common.config import Settings +from app.common.storage import StorageClient +from conftest import SECRET + +HEADER = 'X-LLMiser-Deadline-Ms' +def headers(seconds=60): + return {'X-Internal-Secret':SECRET, HEADER:str(int((time.time()+seconds)*1000))} + +@contextmanager +def active(seconds=60): + value = budget.Budget(time.monotonic()+seconds) + token = budget._current.set(value) + try: yield value + finally: budget._current.reset(token) + +def expire(value): + object.__setattr__(value,'end',time.monotonic()-1) + +@pytest.mark.parametrize('raw',['','abc','-1','1.5','1e13','9'*17]) +def test_invalid_deadline_is_authenticated_before_validation(client,raw): + form = {'files':{'file':('a.txt',b'complete text')}} + response = client.post('/api/v1/internal/convert',headers={'X-Internal-Secret':'wrong',HEADER:raw},**form) + assert response.status_code == 401 + response = client.post('/api/v1/internal/convert',headers={'X-Internal-Secret':SECRET,HEADER:raw},**form) + assert response.status_code == 400 + +def test_expired_deadline_does_not_call_converter(client,monkeypatch): + monkeypatch.setattr(client.app.state.converter,'convert',lambda *a,**k:pytest.fail('expired parser started')) + result = client.post('/api/v1/internal/convert',headers=headers(-1),files={'file':('a.txt',b'text')}) + assert result.status_code == 504 + assert result.headers['X-LLMiser-Deadline-Exceeded'] == '1' + +def test_missing_deadline_and_valid_deadline_produce_identical_complete_results(client): + results = [client.post('/api/v1/internal/convert',headers=h,files={'file':('a.txt','complete café 日本語 🙂'.encode())}).json() for h in [{'X-Internal-Secret':SECRET},headers()]] + for key in ['content','filename','word_count','tokens_est']: + assert results[0][key] == results[1][key] + assert budget.current_budget() is None + +def test_parse_consuming_budget_cannot_write_or_report_success(): + writes=[] + def convert(*args,**kwargs): + assert budget.current_budget() is not None # ASGI → sync worker propagation + expire(budget.current_budget()) + return 'complete output',None + engine=SimpleNamespace(engine_name='test',extensions={'txt'},validate=lambda *a:None,convert=convert) + store=SimpleNamespace(download=lambda *a:b'text',upload=lambda *a:writes.append(a)) + settings=Settings(role='normal',instance='test',shared_secret=SECRET,max_upload_bytes=100,version='test') + client=TestClient(create_app(settings,engine,storage=store)) + job=str(uuid.uuid4()) + response=client.post('/api/v1/internal/process',headers=headers(),json={'job_id':job,'input_path':f'jobs/{job}/input/source.txt','output_path':f'jobs/{job}/output/result.md','original_filename':'a.txt'}) + assert response.status_code == 504 + assert writes == [] + +def test_upload_workers_inherit_budget_and_do_not_leak_it(): + with active() as value, ThreadPoolExecutor(max_workers=2) as pool: + assert list(pool.map(budget.budgeted_worker(lambda n:budget.current_budget()),range(6))) == [value]*6 + expire(value) + with pytest.raises(budget.DeadlineExceeded): + list(pool.map(budget.budgeted_worker(lambda n:pytest.fail('expired upload started')),range(6))) + assert budget.current_budget() is None + + +def test_concurrent_requests_have_independent_budgets(): + barrier=threading.Barrier(2) + seen=[] + def convert(*args,**kwargs): + value=budget.current_budget() + barrier.wait(timeout=5) + seen.append(value.remaining()) + return 'complete',None + engine=SimpleNamespace(engine_name='test',extensions={'txt'},validate=lambda *a:None,convert=convert) + settings=Settings(role='normal',instance='test',shared_secret=SECRET,max_upload_bytes=100,version='test') + client=TestClient(create_app(settings,engine)) + def post(seconds): + return client.post('/api/v1/internal/convert',headers=headers(seconds),files={'file':('a.txt',b'text')}).status_code + with ThreadPoolExecutor(max_workers=2) as pool: + assert list(pool.map(post,[60,120])) == [200,200] + assert min(seen) < 90 < max(seen) + assert budget.current_budget() is None + +def test_storage_timeout_shrinks_and_drip_read_checks_total_budget(): + with active(2) as value: + class Stream(httpx.SyncByteStream): + def __iter__(self): + yield b'first' + expire(value) + yield b'last' + def handler(request): + assert all(0 < seconds <= 2 for seconds in request.extensions['timeout'].values()) + return httpx.Response(200,stream=Stream()) + store=StorageClient('https://storage.test','test','test',transport=httpx.MockTransport(handler)) + with pytest.raises(budget.DeadlineExceeded): + store.download(f'jobs/{uuid.uuid4()}/input/source.txt',100) + + +def test_slow_upload_deadline_is_not_reported_as_storage_success(): + with active(60) as value: + def handler(request): + expire(value) + return httpx.Response(200,json={}) + store=StorageClient('https://storage.test','test','test',transport=httpx.MockTransport(handler)) + with pytest.raises(budget.DeadlineExceeded): + store.upload(f'jobs/{uuid.uuid4()}/output/result.md',b'complete','text/markdown') + + +def test_deadline_io_timeout_retains_deadline_marker_instead_of_peer_failover(): + with active(60) as value: + def handler(request): + expire(value) + raise httpx.ReadTimeout('generated deadline timeout',request=request) + store=StorageClient('https://storage.test','test','test',transport=httpx.MockTransport(handler)) + with pytest.raises(budget.DeadlineExceeded): + store.download(f'jobs/{uuid.uuid4()}/input/source.txt',100) + +def test_far_future_budget_is_clamped_and_duplicate_headers_rejected(client,monkeypatch): + def convert(*a,**k): + assert 0 < budget.current_budget().remaining() <= 300 + return 'complete',None + monkeypatch.setattr(client.app.state.converter,'convert',convert) + result=client.post('/api/v1/internal/convert',headers={'X-Internal-Secret':SECRET,HEADER:'9'*16},files={'file':('a.txt',b'text')}) + assert result.status_code == 200 + result=client.post('/api/v1/internal/convert',headers=[('X-Internal-Secret',SECRET),(HEADER,'1'),(HEADER,'2')],files={'file':('a.txt',b'text')}) + assert result.status_code == 400 diff --git a/backendN/tests/test_pdf_tasks.py b/backendN/tests/test_pdf_tasks.py index e15982a..7f01e72 100644 --- a/backendN/tests/test_pdf_tasks.py +++ b/backendN/tests/test_pdf_tasks.py @@ -65,6 +65,22 @@ def job_paths(): return job, f"jobs/{job}/input/source.pdf", f"jobs/{job}/output/result.md" +def test_deadline_after_pdf_classification_does_not_write_partial_success(client,storage,monkeypatch): + import time + from app.common.deadlines import current_budget + original = pdf_tasks.classify_pages + def classify(pdf): + pages = original(pdf) + object.__setattr__(current_budget(),'end',time.monotonic()-1) + return pages + monkeypatch.setattr(pdf_tasks,'classify_pages',classify) + job,src,out = job_paths() + storage.objects[src] = MIXED + response = client.post('/api/v1/internal/pdf/analyze',headers={**HEADERS,'X-LLMiser-Deadline-Ms':str(int((time.time()+60)*1000))},json={'job_id':job,'input_path':src,'output_path':out,'original_filename':'mixed.pdf'}) + assert response.status_code == 504 + assert set(storage.objects) == {src} + + def test_pages_are_classified_and_grouped(): pdf = pdfium.PdfDocument(MIXED) try: diff --git a/backendO/app/common/app_factory.py b/backendO/app/common/app_factory.py index 08f92f8..a0c5bda 100644 --- a/backendO/app/common/app_factory.py +++ b/backendO/app/common/app_factory.py @@ -26,6 +26,7 @@ from .config import Settings, deployment_revision from .errors import ConversionRejected, EngineUnavailable +from .deadlines import DeadlineExceeded, DeadlineMiddleware, check_budget, compute_checkpoint from .files import read_limited, sanitize_filename, split_name, title_from_stem from .profiles import PROFILES, count_words, finalize_markdown from .security import require_internal_secret @@ -75,6 +76,7 @@ def create_app(settings: Settings, converter: Converter, storage: StorageClient # No interactive docs or schema: these instances are internal compute, # not a public API surface. app = FastAPI(title=f"LLMiser backend ({settings.role})", docs_url=None, redoc_url=None, openapi_url=None) + app.add_middleware(DeadlineMiddleware) app.state.settings = settings app.state.converter = converter app.state.storage = storage if storage is not None else StorageClient.from_env() @@ -95,9 +97,12 @@ def run(original_name: str, profile: str, job_id: str, load_bytes, *, raw: bool if profile not in PROFILES: profile = "Standard" data = load_bytes() + compute_checkpoint() converter.validate(data, ext) title = title_from_stem(stem) + compute_checkpoint() markdown, warning = converter.convert(data, ext, filename=original_name, title=title) + compute_checkpoint() if raw: content = markdown.strip() + "\n" words = count_words(content) @@ -112,6 +117,9 @@ def run(original_name: str, profile: str, job_id: str, load_bytes, *, raw: bool profile=profile, ) outcome = "ok" + except DeadlineExceeded: + outcome = "deadline" + raise except ConversionRejected as err: outcome = err.code raise HTTPException(status_code=err.status_code, detail=err.detail) from None @@ -214,9 +222,11 @@ def process(req: ProcessRequest) -> dict: ) write_started = time.perf_counter() + check_budget() body = content.encode("utf-8") try: storage.upload(req.output_path, body, "text/markdown; charset=utf-8") + check_budget() except StorageError as err: # The conversion worked but the result isn't stored: never report # success. Transient → the dispatcher may retry on the peer. @@ -248,6 +258,7 @@ def convert( job_id, lambda: read_limited(file.file, settings.max_upload_bytes), ) + check_budget() return {**fields, "content": content} return app diff --git a/backendO/app/common/deadlines.py b/backendO/app/common/deadlines.py new file mode 100644 index 0000000..116ae31 --- /dev/null +++ b/backendO/app/common/deadlines.py @@ -0,0 +1,91 @@ +"""Optional authenticated per-request budget; legacy callers keep their limits.""" +from contextvars import ContextVar +from dataclasses import dataclass +from functools import wraps +import re +import time + +from fastapi import HTTPException + +HEADER = b"x-llmiser-deadline-ms" +FINALIZE_RESERVE_S = 10.0 +MAX_BUDGET_S = 300.0 +_current = ContextVar("processing_deadline", default=None) + + +class DeadlineExceeded(HTTPException): + def __init__(self): + super().__init__(504, "Conversion timed out. Try a smaller file or retry later.", + headers={"X-LLMiser-Deadline-Exceeded": "1"}) + + +@dataclass(frozen=True) +class Budget: + end: float + + def remaining(self): + return max(0.0, self.end - time.monotonic()) + + +class DeadlineMiddleware: + """Set context before body parsing/thread admission; auth validates it later.""" + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + value = None + if scope["type"] == "http" and scope.get("path", "").startswith("/api/v1/internal/"): + headers = [v for k, v in scope.get("headers", []) if k.lower() == HEADER] + if headers: + raw = headers[0] + if len(headers) != 1 or not re.fullmatch(rb"[0-9]{1,16}", raw): + value = "invalid" + else: + seconds = min(MAX_BUDGET_S, max(0.0, int(raw) / 1000 - time.time())) + value = Budget(time.monotonic() + seconds) + token = _current.set(value) + try: + await self.app(scope, receive, send) + finally: + _current.reset(token) + + +def current_budget(): + value = _current.get() + if value == "invalid": + raise HTTPException(400, "Invalid processing deadline") + return value + + +def check_budget(reserve=0.0): + budget = current_budget() + if budget is not None and budget.remaining() <= reserve: + raise DeadlineExceeded() + + +def compute_checkpoint(): + check_budget(FINALIZE_RESERVE_S) + + +def timeout_seconds(default, *, reserve=0.0): + budget = current_budget() + if budget is None: + return default + left = budget.remaining() - reserve + if left <= 0: + raise DeadlineExceeded() + return min(default, left) + + +def budgeted_worker(fn): + """ThreadPoolExecutor does not inherit the request's ContextVar.""" + value = _current.get() + @wraps(fn) + def run(*args, **kwargs): + token = _current.set(value) + try: + check_budget() + return fn(*args, **kwargs) + finally: + _current.reset(token) + return run diff --git a/backendO/app/common/security.py b/backendO/app/common/security.py index f1213db..bc36944 100644 --- a/backendO/app/common/security.py +++ b/backendO/app/common/security.py @@ -5,6 +5,7 @@ import hmac from fastapi import Header, HTTPException, Request +from .deadlines import check_budget def require_internal_secret( @@ -23,3 +24,4 @@ def require_internal_secret( x_internal_secret.encode(), expected.encode() ): raise HTTPException(status_code=401, detail="Unauthorized") + check_budget() diff --git a/backendO/app/common/storage.py b/backendO/app/common/storage.py index da6b48c..584e4d4 100644 --- a/backendO/app/common/storage.py +++ b/backendO/app/common/storage.py @@ -16,6 +16,7 @@ from urllib.parse import quote import httpx +from .deadlines import check_budget, timeout_seconds # Backends only touch the fixed object layout of a job (docs/BACKGROUND_JOBS.md). # Anything else is refused, so a bug or a compromised caller can't read or @@ -54,6 +55,7 @@ class StorageClient: def __init__(self, base_url: str, service_key: str, bucket: str, *, timeout: float = 30.0, transport=None): self._base = base_url.rstrip("/") + "/storage/v1" self._bucket = bucket + self._timeout_s = timeout self._http = httpx.Client( timeout=timeout, transport=transport, @@ -73,19 +75,21 @@ def _url(self, path: str) -> str: def download(self, path: str, max_bytes: int) -> bytes: try: - with self._http.stream("GET", self._url(path)) as res: + with self._http.stream("GET", self._url(path), timeout=timeout_seconds(self._timeout_s)) as res: if res.status_code in (400, 404): raise StorageError("Input object not found", transient=False, status=res.status_code) if res.status_code >= 400: raise StorageError(f"Storage read failed ({res.status_code})", transient=res.status_code >= 500, status=res.status_code) chunks, size = [], 0 for chunk in res.iter_bytes(): + check_budget() size += len(chunk) if size > max_bytes: raise StorageError("Input object exceeds the size limit", transient=False, status=413) chunks.append(chunk) return b"".join(chunks) except httpx.HTTPError as err: + check_budget() raise StorageError(f"Storage unreachable: {type(err).__name__}", transient=True) from None def upload(self, path: str, data: bytes, content_type: str) -> None: @@ -94,9 +98,12 @@ def upload(self, path: str, data: bytes, content_type: str) -> None: res = self._http.post( self._url(path), content=data, + timeout=timeout_seconds(self._timeout_s), headers={"Content-Type": content_type, "x-upsert": "true", "Cache-Control": "no-store"}, ) except httpx.HTTPError as err: + check_budget() raise StorageError(f"Storage unreachable: {type(err).__name__}", transient=True) from None if res.status_code >= 400: raise StorageError(f"Storage write failed ({res.status_code})", transient=res.status_code >= 500, status=res.status_code) + check_budget() diff --git a/backendO/app/converter.py b/backendO/app/converter.py index 318e972..d7f828f 100644 --- a/backendO/app/converter.py +++ b/backendO/app/converter.py @@ -22,6 +22,7 @@ from PIL import Image, ImageOps, JpegImagePlugin, UnidentifiedImageError from app.common.errors import ConversionRejected, EngineUnavailable +from app.common.deadlines import FINALIZE_RESERVE_S, compute_checkpoint, timeout_seconds PIL_FORMAT_BY_EXT = { "jpg": "JPEG", @@ -269,7 +270,7 @@ def _ocr(self, img: Image.Image, deadline: float) -> tuple[str, float, int]: img, lang=OCR_LANG, config=f"--oem 1 --psm {OCR_PSM}", - timeout=max(1, int(remaining)), + timeout=max(0.001, min(remaining, timeout_seconds(OCR_TIMEOUT_S, reserve=FINALIZE_RESERVE_S))), output_type=pytesseract.Output.DICT, ) return _text_from_data(data) @@ -287,7 +288,7 @@ def _osd_rotation(self, img: Image.Image) -> int | None: img, config="--psm 0 -c min_characters_to_try=5", output_type=pytesseract.Output.DICT, - timeout=10, + timeout=timeout_seconds(10, reserve=FINALIZE_RESERVE_S), ) except (pytesseract.TesseractError, RuntimeError): return None # too little text, or OSD timed out @@ -301,7 +302,7 @@ def _recognize(self, img: Image.Image) -> tuple[str, int]: decide, the other three orientations) is kept only if it reads with clearly higher confidence. """ - deadline = time.monotonic() + OCR_TIMEOUT_S + deadline = time.monotonic() + timeout_seconds(OCR_TIMEOUT_S, reserve=FINALIZE_RESERVE_S) best_text, best_conf, best_words = self._ocr(img, deadline) best_rotation = 0 proposed = None @@ -315,6 +316,7 @@ def _recognize(self, img: Image.Image) -> tuple[str, int]: candidates = [90, 180, 270] for rotation in candidates: + compute_checkpoint() if time.monotonic() >= deadline: break text, conf, words = self._ocr(_rotate_upright(img, rotation), deadline) @@ -338,6 +340,7 @@ def convert(self, data: bytes, ext: str, *, filename: str, title: str) -> tuple[ # Hold capacity before decoding: otherwise waiting requests retain # their own full-size prepared images while another job runs OCR. with _OCR_LOCK: + compute_checkpoint() img = _open_checked(data, ext) try: frames = getattr(img, "n_frames", 1) @@ -349,16 +352,19 @@ def convert(self, data: bytes, ext: str, *, filename: str, title: str) -> tuple[ img.close() try: + compute_checkpoint() raw, rotation = self._recognize(prepared) except pytesseract.TesseractNotFoundError as err: raise EngineUnavailable(str(err)) from None except RuntimeError as err: + compute_checkpoint() if "timeout" in str(err).lower(): raise ConversionRejected(422, "OCR took too long for this image", "ocr_timeout") from None raise finally: prepared.close() + compute_checkpoint() text = _clean_text(raw) if not text: return "_No text detected in this image._", "no-text-detected" diff --git a/backendO/tests/test_deadlines.py b/backendO/tests/test_deadlines.py new file mode 100644 index 0000000..a96c878 --- /dev/null +++ b/backendO/tests/test_deadlines.py @@ -0,0 +1,47 @@ +from contextlib import contextmanager +import time +import pytest +from app import converter +from app.common import deadlines +from conftest import image_bytes + +@contextmanager +def active(seconds): + value=deadlines.Budget(time.monotonic()+seconds) + token=deadlines._current.set(value) + try: yield value + finally: deadlines._current.reset(token) + +def test_waiting_expired_ocr_does_not_decode(monkeypatch): + monkeypatch.setattr(converter,'_open_checked',lambda *a:pytest.fail('expired OCR decoded image')) + with active(-1),pytest.raises(deadlines.DeadlineExceeded): + converter.OcrConverter().convert(image_bytes('PNG'),'png',filename='a.png',title='a') + +def test_tesseract_timeout_preserves_result_save_reserve(monkeypatch): + import pytesseract + calls=[] + def image_to_data(image,**kwargs): + calls.append(kwargs['timeout']) + return {'text':[],'conf':[],'block_num':[],'par_num':[],'line_num':[]} + monkeypatch.setattr(pytesseract,'image_to_data',image_to_data) + with active(15): + converter.OcrConverter()._ocr(None,time.monotonic()+85) + assert 0 < calls[0] <= 5 + +def test_ocr_budget_timeout_is_504_and_prepared_pixels_close(monkeypatch): + prepared=[] + original=converter._prepare + def prepare(source): + image=original(source) + prepared.append(image) + return image + def recognize(*args): + object.__setattr__(deadlines.current_budget(),'end',time.monotonic()-1) + raise RuntimeError('Tesseract process timeout') + engine=converter.OcrConverter() + monkeypatch.setattr(converter,'_prepare',prepare) + monkeypatch.setattr(engine,'_recognize',recognize) + with active(60),pytest.raises(deadlines.DeadlineExceeded): + engine.convert(image_bytes('PNG'),'png',filename='a.png',title='a') + with pytest.raises(ValueError,match='closed'): + prepared[0].getpixel((0,0)) diff --git a/backendZ/app/archive.py b/backendZ/app/archive.py index 32fac8e..3892156 100644 --- a/backendZ/app/archive.py +++ b/backendZ/app/archive.py @@ -23,6 +23,7 @@ from dataclasses import asdict, dataclass, field from app.common.errors import ConversionRejected +from app.common.deadlines import DeadlineExceeded, compute_checkpoint def _env_int(name: str, default: int) -> int: @@ -221,6 +222,7 @@ def process_archive(data: bytes, convert_document, *, ocr_available: bool) -> li def _walk(archive, infos, prefix, depth, state, out, convert_document, ocr_available): for info in sorted(infos, key=lambda i: i.filename.casefold()): + compute_checkpoint() state.entries_seen += 1 rel = normalize(info.filename) if rel is None: @@ -312,6 +314,7 @@ def _walk(archive, infos, prefix, depth, state, out, convert_document, ocr_avail continue text = _convert(entry, raw, convert_document) + compute_checkpoint() if text is None: continue size = len(text.encode("utf-8")) @@ -327,6 +330,8 @@ def _convert(entry: Entry, raw: bytes, convert_document) -> str | None: if entry.node_type == "DOCUMENT": try: markdown = (convert_document(raw, ext, entry.path.rsplit("/", 1)[-1]) or "").strip() + except DeadlineExceeded: + raise except Exception as err: # noqa: BLE001 - one bad file must not fail the archive entry.status, entry.skip_reason = "FAILED", f"could not be converted ({type(err).__name__})" return None diff --git a/backendZ/app/archive_tasks.py b/backendZ/app/archive_tasks.py index 73453ba..3014988 100644 --- a/backendZ/app/archive_tasks.py +++ b/backendZ/app/archive_tasks.py @@ -32,6 +32,7 @@ from app.common.files import sanitize_filename, split_name, title_from_stem from app.common.profiles import PROFILES, finalize_markdown from app.common.security import require_internal_secret +from app.common.deadlines import DeadlineExceeded, budgeted_worker, check_budget, compute_checkpoint from app.common.storage import StorageError logger = logging.getLogger("mdify.archive") @@ -115,6 +116,7 @@ def partial_blobs(entries: list[archive.Entry]) -> list[bytes]: def write_outputs(store, *, job_id: str, name: str, profile: str, source_bytes: int, entries: list[archive.Entry], ocr_texts: dict[str, str | None], settings) -> dict: + compute_checkpoint() if len(entries) > archive.MAX_ENTRIES: raise resource_limit() stem, _ext = split_name(name) @@ -154,10 +156,12 @@ def write_outputs(store, *, job_id: str, name: str, profile: str, source_bytes: raise resource_limit() try: + check_budget() with ThreadPoolExecutor(max_workers=WORKERS) as pool: # Encoding inside the worker bounds live application byte buffers # to WORKERS objects instead of retaining every encoded part. - list(pool.map(lambda f: store.upload(f[0], f[1].encode("utf-8"), f[2]), files)) + list(pool.map(budgeted_worker(lambda f: store.upload(f[0], f[1].encode("utf-8"), f[2])), files)) + check_budget() except StorageError as err: raise HTTPException(status_code=503 if err.transient else 502, detail="Could not store the project result") from None @@ -208,6 +212,8 @@ def process(req: ProcessRequest) -> dict: raise HTTPException(status_code=err.status_code, detail=err.detail) from None except EngineUnavailable: raise HTTPException(status_code=503, detail="Conversion engine unavailable") from None + except DeadlineExceeded: + raise except Exception: logger.exception("archive processing failed job=%s", job_id) raise HTTPException(status_code=422, detail="This ZIP file couldn't be processed.") from None @@ -244,6 +250,7 @@ def process(req: ProcessRequest) -> dict: # Converted entries wait in Storage for the merge, in parts <= 4 MB. parts = partial_blobs(entries) + check_budget() partial_paths = [f"jobs/{job_id}/nodes/archive-{i}/partial.json" for i in range(1, len(parts) + 1)] uploads += [ (path, chunk, "application/json; charset=utf-8") @@ -251,7 +258,8 @@ def process(req: ProcessRequest) -> dict: ] try: with ThreadPoolExecutor(max_workers=WORKERS) as pool: - list(pool.map(lambda u: store.upload(*u), uploads)) + list(pool.map(budgeted_worker(lambda u: store.upload(*u)), uploads)) + check_budget() except StorageError as err: raise HTTPException(status_code=503 if err.transient else 502, detail="Could not store the archive contents") from None @@ -281,6 +289,7 @@ def merge(req: MergeRequest) -> dict: def fetch(path: str) -> str | None: nonlocal read_bytes + compute_checkpoint() try: data = store.download(path, PARTIAL_BYTES) except StorageError as err: diff --git a/backendZ/app/common/app_factory.py b/backendZ/app/common/app_factory.py index 08f92f8..a0c5bda 100644 --- a/backendZ/app/common/app_factory.py +++ b/backendZ/app/common/app_factory.py @@ -26,6 +26,7 @@ from .config import Settings, deployment_revision from .errors import ConversionRejected, EngineUnavailable +from .deadlines import DeadlineExceeded, DeadlineMiddleware, check_budget, compute_checkpoint from .files import read_limited, sanitize_filename, split_name, title_from_stem from .profiles import PROFILES, count_words, finalize_markdown from .security import require_internal_secret @@ -75,6 +76,7 @@ def create_app(settings: Settings, converter: Converter, storage: StorageClient # No interactive docs or schema: these instances are internal compute, # not a public API surface. app = FastAPI(title=f"LLMiser backend ({settings.role})", docs_url=None, redoc_url=None, openapi_url=None) + app.add_middleware(DeadlineMiddleware) app.state.settings = settings app.state.converter = converter app.state.storage = storage if storage is not None else StorageClient.from_env() @@ -95,9 +97,12 @@ def run(original_name: str, profile: str, job_id: str, load_bytes, *, raw: bool if profile not in PROFILES: profile = "Standard" data = load_bytes() + compute_checkpoint() converter.validate(data, ext) title = title_from_stem(stem) + compute_checkpoint() markdown, warning = converter.convert(data, ext, filename=original_name, title=title) + compute_checkpoint() if raw: content = markdown.strip() + "\n" words = count_words(content) @@ -112,6 +117,9 @@ def run(original_name: str, profile: str, job_id: str, load_bytes, *, raw: bool profile=profile, ) outcome = "ok" + except DeadlineExceeded: + outcome = "deadline" + raise except ConversionRejected as err: outcome = err.code raise HTTPException(status_code=err.status_code, detail=err.detail) from None @@ -214,9 +222,11 @@ def process(req: ProcessRequest) -> dict: ) write_started = time.perf_counter() + check_budget() body = content.encode("utf-8") try: storage.upload(req.output_path, body, "text/markdown; charset=utf-8") + check_budget() except StorageError as err: # The conversion worked but the result isn't stored: never report # success. Transient → the dispatcher may retry on the peer. @@ -248,6 +258,7 @@ def convert( job_id, lambda: read_limited(file.file, settings.max_upload_bytes), ) + check_budget() return {**fields, "content": content} return app diff --git a/backendZ/app/common/deadlines.py b/backendZ/app/common/deadlines.py new file mode 100644 index 0000000..116ae31 --- /dev/null +++ b/backendZ/app/common/deadlines.py @@ -0,0 +1,91 @@ +"""Optional authenticated per-request budget; legacy callers keep their limits.""" +from contextvars import ContextVar +from dataclasses import dataclass +from functools import wraps +import re +import time + +from fastapi import HTTPException + +HEADER = b"x-llmiser-deadline-ms" +FINALIZE_RESERVE_S = 10.0 +MAX_BUDGET_S = 300.0 +_current = ContextVar("processing_deadline", default=None) + + +class DeadlineExceeded(HTTPException): + def __init__(self): + super().__init__(504, "Conversion timed out. Try a smaller file or retry later.", + headers={"X-LLMiser-Deadline-Exceeded": "1"}) + + +@dataclass(frozen=True) +class Budget: + end: float + + def remaining(self): + return max(0.0, self.end - time.monotonic()) + + +class DeadlineMiddleware: + """Set context before body parsing/thread admission; auth validates it later.""" + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + value = None + if scope["type"] == "http" and scope.get("path", "").startswith("/api/v1/internal/"): + headers = [v for k, v in scope.get("headers", []) if k.lower() == HEADER] + if headers: + raw = headers[0] + if len(headers) != 1 or not re.fullmatch(rb"[0-9]{1,16}", raw): + value = "invalid" + else: + seconds = min(MAX_BUDGET_S, max(0.0, int(raw) / 1000 - time.time())) + value = Budget(time.monotonic() + seconds) + token = _current.set(value) + try: + await self.app(scope, receive, send) + finally: + _current.reset(token) + + +def current_budget(): + value = _current.get() + if value == "invalid": + raise HTTPException(400, "Invalid processing deadline") + return value + + +def check_budget(reserve=0.0): + budget = current_budget() + if budget is not None and budget.remaining() <= reserve: + raise DeadlineExceeded() + + +def compute_checkpoint(): + check_budget(FINALIZE_RESERVE_S) + + +def timeout_seconds(default, *, reserve=0.0): + budget = current_budget() + if budget is None: + return default + left = budget.remaining() - reserve + if left <= 0: + raise DeadlineExceeded() + return min(default, left) + + +def budgeted_worker(fn): + """ThreadPoolExecutor does not inherit the request's ContextVar.""" + value = _current.get() + @wraps(fn) + def run(*args, **kwargs): + token = _current.set(value) + try: + check_budget() + return fn(*args, **kwargs) + finally: + _current.reset(token) + return run diff --git a/backendZ/app/common/security.py b/backendZ/app/common/security.py index f1213db..bc36944 100644 --- a/backendZ/app/common/security.py +++ b/backendZ/app/common/security.py @@ -5,6 +5,7 @@ import hmac from fastapi import Header, HTTPException, Request +from .deadlines import check_budget def require_internal_secret( @@ -23,3 +24,4 @@ def require_internal_secret( x_internal_secret.encode(), expected.encode() ): raise HTTPException(status_code=401, detail="Unauthorized") + check_budget() diff --git a/backendZ/app/common/storage.py b/backendZ/app/common/storage.py index da6b48c..584e4d4 100644 --- a/backendZ/app/common/storage.py +++ b/backendZ/app/common/storage.py @@ -16,6 +16,7 @@ from urllib.parse import quote import httpx +from .deadlines import check_budget, timeout_seconds # Backends only touch the fixed object layout of a job (docs/BACKGROUND_JOBS.md). # Anything else is refused, so a bug or a compromised caller can't read or @@ -54,6 +55,7 @@ class StorageClient: def __init__(self, base_url: str, service_key: str, bucket: str, *, timeout: float = 30.0, transport=None): self._base = base_url.rstrip("/") + "/storage/v1" self._bucket = bucket + self._timeout_s = timeout self._http = httpx.Client( timeout=timeout, transport=transport, @@ -73,19 +75,21 @@ def _url(self, path: str) -> str: def download(self, path: str, max_bytes: int) -> bytes: try: - with self._http.stream("GET", self._url(path)) as res: + with self._http.stream("GET", self._url(path), timeout=timeout_seconds(self._timeout_s)) as res: if res.status_code in (400, 404): raise StorageError("Input object not found", transient=False, status=res.status_code) if res.status_code >= 400: raise StorageError(f"Storage read failed ({res.status_code})", transient=res.status_code >= 500, status=res.status_code) chunks, size = [], 0 for chunk in res.iter_bytes(): + check_budget() size += len(chunk) if size > max_bytes: raise StorageError("Input object exceeds the size limit", transient=False, status=413) chunks.append(chunk) return b"".join(chunks) except httpx.HTTPError as err: + check_budget() raise StorageError(f"Storage unreachable: {type(err).__name__}", transient=True) from None def upload(self, path: str, data: bytes, content_type: str) -> None: @@ -94,9 +98,12 @@ def upload(self, path: str, data: bytes, content_type: str) -> None: res = self._http.post( self._url(path), content=data, + timeout=timeout_seconds(self._timeout_s), headers={"Content-Type": content_type, "x-upsert": "true", "Cache-Control": "no-store"}, ) except httpx.HTTPError as err: + check_budget() raise StorageError(f"Storage unreachable: {type(err).__name__}", transient=True) from None if res.status_code >= 400: raise StorageError(f"Storage write failed ({res.status_code})", transient=res.status_code >= 500, status=res.status_code) + check_budget() diff --git a/backendZ/tests/test_deadlines.py b/backendZ/tests/test_deadlines.py new file mode 100644 index 0000000..890e1c8 --- /dev/null +++ b/backendZ/tests/test_deadlines.py @@ -0,0 +1,31 @@ +import time +import pytest +from app import archive +from app.common.deadlines import current_budget +from test_archive_tasks import HEADERS,PROCESS,process_body,upload + +def test_archive_expiry_is_not_hidden_as_one_bad_document(client,storage,converter,monkeypatch): + original = converter.documents.__class__.__call__ + calls=[] + def convert(self,*args): + calls.append(True) + text=original(self,*args) + object.__setattr__(current_budget(),'end',time.monotonic()-1) + return text + monkeypatch.setattr(converter.documents.__class__,'__call__',convert) + job=upload(storage,{'first.html':'

Complete first

','second.html':'

Complete second

'}) + response=client.post(PROCESS,headers={**HEADERS,'X-LLMiser-Deadline-Ms':str(int((time.time()+60)*1000))},json=process_body(job)) + assert response.status_code == 504 + assert len(calls) == 1 + assert not any('/output/' in name for name in storage.objects) + +@pytest.mark.parametrize('endpoint',[PROCESS,'/api/v1/internal/archive/merge']) +def test_archive_expired_requests_fail_before_storage(client,storage,endpoint): + job=upload(storage,{'note.txt':'complete'}) + body=process_body(job) + if endpoint.endswith('merge'): + body.update(partial_paths=[f'jobs/{job}/nodes/one/partial.json'],ocr=[]) + previous=dict(storage.objects) + response=client.post(endpoint,headers={**HEADERS,'X-LLMiser-Deadline-Ms':'1'},json=body) + assert response.status_code == 504 + assert storage.objects == previous diff --git a/docs/LLMISER_OPTIMIZATION_PLAN.md b/docs/LLMISER_OPTIMIZATION_PLAN.md index ea15a0b..167e7ce 100644 --- a/docs/LLMISER_OPTIMIZATION_PLAN.md +++ b/docs/LLMISER_OPTIMIZATION_PLAN.md @@ -14,6 +14,8 @@ PR #13 released [a source capacity audit and CSV allocation reduction](LLMISER_P ## Verified starting point +Current candidate: [shared conversion deadlines](LLMISER_SHARED_DEADLINES.md), branch `codex/shared-conversion-deadlines`. Backend cooperative checkpoints and remaining-budget propagation are implemented locally; rollout/acceptance remain separate. This is a per-invocation budget and adds no whole-job lifetime, SQL or capacity setting. + Released [Excel and final-result assembly memory](LLMISER_EXCEL_ASSEMBLY_MEMORY.md), PR #15 at `3db41140521d5cfb1bb3bfd12865feca5a8cc95b`: sequential sheet processing, exact bounded word counts and delayed archive output encoding. Three fresh-process trials per mode preserve complete hashes; local peak reductions are qualified by timing variability/XLS regression. All seven CI jobs passed before/after merge; six production Vercel statuses succeeded; all six N/O/Z workers report the exact revision and readiness. Eight generated multipart probes, two production Storage conversion/recovery checks and 20 HTTP checks passed; all pools are 2/2 ready. No SQL, concurrency, retention or blog changes. The copied source already contains atomic job startup/finalization, cleanup recovery, upload/active-job/storage-reservation admission, shared archive container checks, bounded PDF page uploads/OCR preparation, lazy Markdown/highlighter loading and current-edit export snapshots. Do not reimplement these based on old “proposed” checklist rows. Their deployment/capacity evidence is distinct from source tests. diff --git a/docs/LLMISER_SHARED_DEADLINES.md b/docs/LLMISER_SHARED_DEADLINES.md new file mode 100644 index 0000000..9886f7b --- /dev/null +++ b/docs/LLMISER_SHARED_DEADLINES.md @@ -0,0 +1,29 @@ +# Shared conversion deadlines + +Updated 29 October 2026. Candidate on `codex/shared-conversion-deadlines`; release evidence is pending. The preceding [Excel/assembly release](LLMISER_EXCEL_ASSEMBLY_MEMORY.md) remains the last verified runtime snapshot. + +## Scope and behavior + +Conversion, start, advance and cron tick routes establish one 300-second invocation deadline at entry. Admission, upload/body parsing and queue setup consume that same time. This is **not a new whole-job lifetime limit**: durable PDF/OCR/archive children and later ticks receive their next invocation's budget. Existing leases, attempt limits, admission, retention, SQL and pool capacity are unchanged. + +The dispatcher keeps the existing per-attempt caps (120 seconds normal/archive, 150 seconds OCR), but peer retries share the outer remaining time. It stops backend work 60 seconds before the route deadline to leave time for database outcome recording and result signing. A new attempt requires at least 13 seconds. Backend headers expire two seconds before the transport timer, leaving a response margin. Response-body reading/decoding also consumes the attempt budget. Actual transport timeout or an explicit backend deadline response does not launch the peer during that pass. Legacy transient responses still permit the existing bounded failover when enough time remains. + +Internal protocol: optional `X-LLMiser-Deadline-Ms`, an absolute Unix millisecond timestamp. Absolute time accounts for network transfer, Render wake-up and backend thread admission. Only authenticated internal calls validate it: wrong/missing credentials retain the existing authentication errors. Malformed/duplicate deadline fields return 400; expired authenticated calls return 504 with `X-LLMiser-Deadline-Exceeded: 1`. Missing deadline fields preserve legacy limits. A far-future value is capped at 300 seconds on receipt. Browser-provided deadline headers are not forwarded by the frontend. + +Workers convert the remaining wall-clock time to a monotonic request-local budget at ASGI entry. Context propagates to FastAPI's sync worker; existing `ThreadPoolExecutor` operations explicitly bind/reset the budget. N/O/Z keep byte-identical `common/` sources. PDF loops, archive entries, parser boundaries, OCR preparation/recognition, merge reads and final writes check that budget. An expiry is not swallowed as one bad ZIP document or returned as a successful partial result. + +Compute checkpoints leave ten seconds for worker finalization. OCR and orientation subprocess timeouts shrink to the available compute time. Storage timeouts also shrink, streamed reads check the total deadline between chunks, and writes check again after acknowledgement. An expired output write cannot report successful completion. Database REST calls on start/advance/tick are bounded by both their existing 30-second cap and the outer deadline. Already claimed work follows the existing retry/failure/lease-recovery rules; a failed outcome write is not fabricated as success. + +This is cooperative cancellation. A running native parser, lock wait, network operation or synchronous JSON/Markdown operation cannot be forcibly preempted by these checkpoints. HTTPX's connect/read/write/pool timeouts are inactivity limits, not a hard total-transfer timer; chunk/boundary checks complement them. A response or Storage write may already have reached the server when a timeout is detected. There is no new exactly-once write guarantee, atomic multi-object rollback, concurrency increase or billing claim. Shared provider clocks are assumed accurate; clock skew may cause conservative early failure. Remaining parser isolation, storage attempt fencing and capacity work stay separate. + +References: [Node AbortSignal](https://nodejs.org/api/globals.html#static-method-abortsignaltimeoutdelay), [Python 3.12 context variables](https://docs.python.org/3.12/library/contextvars.html), [AnyIO thread context/cancellation](https://anyio.readthedocs.io/en/stable/threads.html), [HTTPX timeout semantics](https://www.python-httpx.org/advanced/timeouts/). Context7 tools were unavailable; official documentation and installed-runtime tests were used. + +## Validation and rollout + +Local/generated checks currently pass: 270 frontend tests, 160 normal/PDF tests, 104 archive tests and 11 focused OCR/capacity tests. Coverage includes slow primary failover, no-time-before-dispatch, body decoding timeout, old-backend failover, expired database calls, queue claim suppression and durable retry recording; authenticated malformed/expired fields, valid/legacy output equality, clamping, duplicate fields, concurrent context isolation, upload-worker propagation, drip reads, slow upload acknowledgement, PDF/ZIP expiry, OCR subprocess budgets and image cleanup. Existing common-package integrity checks remain intact. The prior workstation-wide OCR readiness-suite limitation remains documented in the Excel report; full Linux OCR CI is required for release. + +Local lint, production build and dictionary/SQL/blog consistency passed. Before release, eight generated multipart workbook/ZIP requests with future deadline headers passed on all four document workers, confirming old-worker compatibility. Two actual public Storage conversions (XLSX and workbook ZIP), all four output downloads and exact saved-result recovery also passed as baselines; those two generated jobs follow normal retention. + +CI, deployment and post-release acceptance are pending. Mixed versions remain compatible during rollout: old workers ignore the additional header, while new workers accept callers without it. Backend cooperative limits are therefore only verified after every worker has the new revision. Do not infer production deployment or savings from local tests. + +No UI/localization/blog/visual, database migration, cron SQL, provider environment, supported-format, file-limit or retention change is included. Operator probes and raw evidence stay ignored under `project-docs/deadline-audit/`. diff --git a/docs/START_HERE.md b/docs/START_HERE.md index 4014ecf..eca3e6a 100644 --- a/docs/START_HERE.md +++ b/docs/START_HERE.md @@ -14,6 +14,8 @@ Released [parser capacity and CSV memory reduction](LLMISER_PARSER_CAPACITY.md) Released [Excel and final-result memory](LLMISER_EXCEL_ASSEMBLY_MEMORY.md) processes workbook sheets sequentially, bounds temporary word-count allocations and defers ZIP output encoding to the existing upload workers. Local untraced peaks fell about 2% for XLSX, 5% for XLS, 15% for ZIP assembly and 17% for profile finalization, with identical output; timing varies and the XLS median regressed. Production output/recovery equivalence passed. Native/concurrent parser peaks, full normalization/serialization, representative-device input behavior and the shared-deadline queue remain separate work. +Current candidate: [shared conversion deadlines](LLMISER_SHARED_DEADLINES.md), branch `codex/shared-conversion-deadlines`. Retries share each invocation's remaining time; workers receive an optional authenticated deadline and reserve finalization time. Existing job lifetime, queue limits and retention stay compatible. Release checks are pending; the runtime revision above remains the last verified release. + ## Legacy frontend replacement is deployed The owner wants the old converter removed from its frontend hosting, permanent public redirects preserved, and a small LLMiser-only banner as fallback. Replacement source is [legacy-site/](../legacy-site/README.md), inside this repository. It has no converter, API or database client. The working LLMiser project continues using `frontend`; only the old frontend project may use `legacy-site`. diff --git a/docs/ai-handover/HANDOVER.md b/docs/ai-handover/HANDOVER.md index d402e7b..06b7de0 100644 --- a/docs/ai-handover/HANDOVER.md +++ b/docs/ai-handover/HANDOVER.md @@ -54,6 +54,8 @@ Active local tooling now has zero absolute references to either old checkout. LL ## Verification commands +Current candidate: [shared conversion deadlines](../LLMISER_SHARED_DEADLINES.md), `codex/shared-conversion-deadlines`. Distinguish its pending release from the verified Excel runtime. The protocol is optional and per invocation, with a 60-second frontend recording/signing reserve and ten-second worker finalization reserve. Preserve legacy callers, auth precedence, context isolation, complete-result semantics and existing durable retry/lease rules. Native parser execution and lock waits remain cooperative rather than forcibly cancelled. + Released optimization: [Excel and final-result memory](../LLMISER_EXCEL_ASSEMBLY_MEMORY.md), PR #15 at 3db4114. N/Z Excel adapters use verified MarkItDown 0.1.8/0.1.9, the same per-sheet pandas/HTML formatting, sequential frame release and XLS on-demand sheet unloading. Shared 64 Ki-character word counting preserves Python whitespace/token semantics across N/O/Z. Archive writes retain text/byte counts and encode within the existing four upload workers after full preflight. Preserve unknown-version/image-hook fallback, complete results and preflight-before-write behavior. Local measurements are not hosting/concurrency guarantees; the small production Storage fixtures do not establish large-job capacity. The owner asked to continue the optimization plan after the retirement handover. PR #8 completed that handover; PR #9 released [shared reader statistics](../LLMISER_READER_STATISTICS.md). PR #11 released [large-preview control and reproducible browser profiling](../LLMISER_LARGE_PREVIEWS.md): above 50,000 UTF-16 characters, formatting waits for an explicit request. Editing/document switching resets the request; complete raw text/copy/exports remain available. All six packs contain 631 messages. Local evidence: 261 full tests, final 20 focused tests, lint/consistency checks and production build passed; generated browser fixtures verify full rendering, edit re-pause, identical-text document switching, manual pause and exact current-edit download. CI passed all seven jobs, production deployed revision 0501cd0, the 20-check HTTP audit passed, and normal/OCR/archive were each 2/2 ready. No fresh production conversion/admin/cleanup acceptance is implied by that audit. diff --git a/docs/ai-handover/PROGRESS.md b/docs/ai-handover/PROGRESS.md index 2ba0661..1f5def1 100644 --- a/docs/ai-handover/PROGRESS.md +++ b/docs/ai-handover/PROGRESS.md @@ -50,6 +50,8 @@ See [release record](../LLMISER_RELEASE_2026-10-26.md), [localization](../LLMISE ## Remaining work +In progress: [shared conversion deadlines](../LLMISER_SHARED_DEADLINES.md). Candidate code and generated tests are complete; production build, CI and deployed protocol/Storage acceptance are pending. This does not replace the released Excel evidence below. + Released: [Excel conversion and final-result assembly memory](../LLMISER_EXCEL_ASSEMBLY_MEMORY.md), explicitly selected by the owner, PR #15 at 3db4114. Local suites: 144 normal/101 archive tests; installed-version checks passed. All seven Linux CI jobs and six Vercel production statuses passed, with exact revision/readiness verified on all six workers including Render. Eight post-release generated multipart probes and two Storage-backed XLSX/ZIP conversions passed, including all four output objects and exact saved-result recovery. Before/after Markdown, counts and index hashes match; manifest job IDs correctly differ. All 20 HTTP checks passed and all pools were 2/2 ready. Four real jobs across before/after tests follow normal retention; cleanup execution itself was not tested. Local Windows OCR readiness-suite failure is preserved in the report; Linux OCR CI passed. The owner resumed the implementation plan after the handover was merged in PR #8. [Shared reader statistics](../LLMISER_READER_STATISTICS.md) shipped in PR #9. [Large-preview control](../LLMISER_LARGE_PREVIEWS.md) shipped in PR #11 at 0501cd0 after 261 frontend CI tests/all seven jobs passed. All six production deployments succeeded; the 28 October 20-check HTTP audit passed and all three pools were 2/2 ready. Actual-component generated browser measurements and exact raw/download acceptance support the preview boundary. Representative-device/native textarea profiling and deeper parser-capacity work remain separate. diff --git a/frontend/app/api/convert/route.js b/frontend/app/api/convert/route.js index f7c61af..5f8da94 100644 --- a/frontend/app/api/convert/route.js +++ b/frontend/app/api/convert/route.js @@ -2,6 +2,7 @@ import { admitPublicRequest, admissionResponse, readBoundedBody } from '../../.. import { NextResponse } from 'next/server'; import { SUPPORTED_EXTENSIONS, maxFileSizeFor, poolForExtension, sizeLimitMessage, splitFileName } from '../../../lib/formats'; import { dispatchConversion, getPoolConfig } from '../../../lib/server/dispatcher'; +import { routeDeadline } from '../../../lib/server/timeBudget'; // Interim transport: the browser still posts the file here and this route // forwards it to a backend. The Storage flow (/api/uploads/create, then @@ -10,6 +11,7 @@ export const dynamic = 'force-dynamic'; export const maxDuration = 300; // Vercel Hobby maximum; see TIMEOUT_MS_BY_POOL in the dispatcher export async function POST(request) { + const deadlineMs = routeDeadline(); const secret = process.env.BACKEND_SHARED_SECRET; if (!secret) { console.error('BACKEND_SHARED_SECRET is not set; refusing to dispatch.'); @@ -64,6 +66,7 @@ export async function POST(request) { file, profile, secret, + deadlineMs, }); releaseLease = status < 500 && attempts === 1; diff --git a/frontend/app/api/jobs/[id]/advance/route.js b/frontend/app/api/jobs/[id]/advance/route.js index fd6ffc6..442a001 100644 --- a/frontend/app/api/jobs/[id]/advance/route.js +++ b/frontend/app/api/jobs/[id]/advance/route.js @@ -2,6 +2,7 @@ import { admitPublicRequest, admissionResponse } from '../../../../../lib/server import { NextResponse } from 'next/server'; import { createSupabase, supabaseConfig } from '../../../../../lib/server/supabaseRest'; import { JobError, advanceJob } from '../../../../../lib/server/jobService'; +import { DeadlineExceededError, routeDeadline } from '../../../../../lib/server/timeBudget'; export const dynamic = 'force-dynamic'; export const maxDuration = 300; // Vercel Hobby maximum; see TIMEOUT_MS_BY_POOL in the dispatcher @@ -11,6 +12,7 @@ const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; // Called repeatedly by the browser while a multi-part job (scanned PDF, // archive) runs: one scheduling pass for this job, then its status. export async function POST(request, { params }) { + const deadlineMs = routeDeadline(); const config = supabaseConfig(); const secret = process.env.BACKEND_SHARED_SECRET; if (!config || !secret) { @@ -22,12 +24,12 @@ export async function POST(request, { params }) { try { await admitPublicRequest(request, 'advance'); - const body = await advanceJob(createSupabase(config), params.id, { secret }); + const body = await advanceJob(createSupabase(config, fetch, { deadlineMs }), params.id, { secret, deadlineMs }); return NextResponse.json(body, { headers: { 'Cache-Control': 'no-store' } }); } catch (err) { const denied = admissionResponse(err); if (denied) return denied; - if (err instanceof JobError) return NextResponse.json({ detail: err.message }, { status: err.status }); + if (err instanceof JobError || err instanceof DeadlineExceededError) return NextResponse.json({ detail: err.message }, { status: err.status }); console.error('[jobs/advance]', err.message); return NextResponse.json({ detail: 'Could not check the conversion. Retrying…' }, { status: 502 }); } diff --git a/frontend/app/api/jobs/[id]/start/route.js b/frontend/app/api/jobs/[id]/start/route.js index 852feda..dfebea5 100644 --- a/frontend/app/api/jobs/[id]/start/route.js +++ b/frontend/app/api/jobs/[id]/start/route.js @@ -2,6 +2,7 @@ import { admitPublicRequest, admissionResponse } from '../../../../../lib/server import { NextResponse } from 'next/server'; import { createSupabase, supabaseConfig } from '../../../../../lib/server/supabaseRest'; import { JobError, startJob } from '../../../../../lib/server/jobService'; +import { DeadlineExceededError, routeDeadline } from '../../../../../lib/server/timeBudget'; export const dynamic = 'force-dynamic'; export const maxDuration = 300; // Vercel Hobby maximum; see TIMEOUT_MS_BY_POOL in the dispatcher @@ -13,6 +14,7 @@ const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; // browser keeps calling /advance. Responses carry statistics and short-lived // download links, never the Markdown itself. export async function POST(request, { params }) { + const deadlineMs = routeDeadline(); const config = supabaseConfig(); const secret = process.env.BACKEND_SHARED_SECRET; if (!config || !secret) { @@ -24,12 +26,12 @@ export async function POST(request, { params }) { try { await admitPublicRequest(request, 'start'); - const { status, body } = await startJob(createSupabase(config), params.id, { secret }); + const { status, body } = await startJob(createSupabase(config, fetch, { deadlineMs }), params.id, { secret, deadlineMs }); return NextResponse.json(body, { status, headers: { 'Cache-Control': 'no-store' } }); } catch (err) { const denied = admissionResponse(err); if (denied) return denied; - if (err instanceof JobError) return NextResponse.json({ detail: err.message }, { status: err.status }); + if (err instanceof JobError || err instanceof DeadlineExceededError) return NextResponse.json({ detail: err.message }, { status: err.status }); console.error('[jobs/start]', err.message); return NextResponse.json({ detail: 'Conversion could not be started. Please retry.' }, { status: 502 }); } diff --git a/frontend/app/api/jobs/tick/route.js b/frontend/app/api/jobs/tick/route.js index 4b8d8ff..ada4d6b 100644 --- a/frontend/app/api/jobs/tick/route.js +++ b/frontend/app/api/jobs/tick/route.js @@ -2,6 +2,7 @@ import { NextResponse } from 'next/server'; import { timingSafeEqual } from 'node:crypto'; import { createSupabase, supabaseConfig } from '../../../../lib/server/supabaseRest'; import { runTick } from '../../../../lib/server/jobQueue'; +import { DeadlineExceededError, routeDeadline } from '../../../../lib/server/timeBudget'; export const dynamic = 'force-dynamic'; export const maxDuration = 300; // Vercel Hobby maximum @@ -17,6 +18,7 @@ function authorized(request) { // retires stale jobs, requeues expired leases and runs one scheduling pass // across all jobs, so work continues when nobody's browser is waiting. export async function POST(request) { + const deadlineMs = routeDeadline(); const config = supabaseConfig(); const secret = process.env.BACKEND_SHARED_SECRET; if (!config || !secret || !process.env.CRON_SECRET) { @@ -24,17 +26,18 @@ export async function POST(request) { } if (!authorized(request)) return NextResponse.json({ detail: 'Unauthorized' }, { status: 401 }); - const db = createSupabase(config); + const db = createSupabase(config, fetch, { deadlineMs }); try { await db.rpc('prune_public_admission', {}); const stale = await db.rpc('sweep_stale_jobs', {}); - const { claimed, results } = await runTick(db, { secret }); + const { claimed, results } = await runTick(db, { secret, deadlineMs }); return NextResponse.json({ stale: Array.isArray(stale) ? stale.length : 0, claimed, outcomes: results.reduce((acc, r) => ({ ...acc, [r.outcome]: (acc[r.outcome] || 0) + 1 }), {}), }); } catch (err) { + if (err instanceof DeadlineExceededError) return NextResponse.json({ detail: err.message }, { status: err.status }); console.error('[jobs/tick]', err.message); return NextResponse.json({ detail: 'Tick failed' }, { status: 502 }); } diff --git a/frontend/lib/server/dispatcher.js b/frontend/lib/server/dispatcher.js index 56db1d2..688a216 100644 --- a/frontend/lib/server/dispatcher.js +++ b/frontend/lib/server/dispatcher.js @@ -8,6 +8,8 @@ // infrastructure failure is retried once on the peer; validation, policy and // conversion failures are never retried. +import { DEADLINE_HEADER, DEADLINE_EXCEEDED_HEADER, FINALIZATION_RESERVE_MS, MIN_ATTEMPT_MS, TIMEOUT_DETAIL, remainingMs, routeDeadline } from './timeBudget.js'; + const TRANSIENT_STATUSES = new Set([502, 503, 504]); // Per-attempt budget. O1/O2 run on Render Free, which sleeps after 15 idle @@ -72,6 +74,8 @@ export async function sendToPool({ secret, timeoutMs = TIMEOUT_MS_BY_POOL[pool] ?? TIMEOUT_MS_BY_POOL.normal, fetchImpl = fetch, + deadlineMs = routeDeadline(), + now = Date.now, }) { const order = attemptOrder(jobId, urls); if (order.length === 0) { @@ -85,35 +89,41 @@ export async function sendToPool({ let lastFailure = null; let attempts = 0; + const stopAt = deadlineMs - FINALIZATION_RESERVE_MS; + const timedOut = (instanceUrl = null) => ({ status: 504, body: { detail: TIMEOUT_DETAIL }, instanceUrl, attempts }); for (const baseUrl of order) { + if (remainingMs(stopAt, now()) < MIN_ATTEMPT_MS) return timedOut(); + const attemptDeadline = Math.min(stopAt, now() + timeoutMs); const init = makeInit(); + const attemptTimeout = remainingMs(attemptDeadline, now()); + if (attemptTimeout < MIN_ATTEMPT_MS) return timedOut(); attempts += 1; let res; try { res = await fetchImpl(`${baseUrl}${path}`, { ...init, method: 'POST', - headers: { ...(init.headers || {}), 'X-Internal-Secret': secret }, - signal: AbortSignal.timeout(timeoutMs), + headers: { ...(init.headers || {}), 'X-Internal-Secret': secret, [DEADLINE_HEADER]: String(attemptDeadline - 2000) }, + signal: AbortSignal.timeout(attemptTimeout), cache: 'no-store', }); } catch (err) { if (err?.name === 'TimeoutError') { // The instance may still be working on it, and a second full attempt // would overrun this function's own time limit — don't retry. - return { - status: 504, - body: { detail: 'Conversion timed out. Try a smaller file or retry later.' }, - instanceUrl: baseUrl, - attempts, - }; + return timedOut(baseUrl); } // Connection refused / DNS / reset — infrastructure, so try the peer. lastFailure = { status: 502, reason: 'unreachable' }; continue; } + if (res.status === 504 && res.headers?.get?.(DEADLINE_EXCEEDED_HEADER) === '1') { + try { await res.body?.cancel?.(); } catch { /* Deadline already terminal for this pass. */ } + return timedOut(baseUrl); + } if (isTransientStatus(res.status)) { + try { await res.body?.cancel?.(); } catch { /* Continue only within the shared budget. */ } lastFailure = { status: res.status, reason: `HTTP ${res.status}` }; continue; } @@ -121,7 +131,9 @@ export async function sendToPool({ let body; try { body = await res.json(); - } catch { + if (remainingMs(attemptDeadline, now()) === 0) return timedOut(baseUrl); + } catch (err) { + if (err?.name === 'TimeoutError' || remainingMs(attemptDeadline, now()) === 0) return timedOut(baseUrl); return { status: res.ok ? 502 : res.status, body: { detail: `Backend returned an unreadable response (${res.status}).` }, instanceUrl: baseUrl, attempts }; } return { status: res.status, body, instanceUrl: baseUrl, attempts }; diff --git a/frontend/lib/server/jobQueue.js b/frontend/lib/server/jobQueue.js index 917061a..79a521f 100644 --- a/frontend/lib/server/jobQueue.js +++ b/frontend/lib/server/jobQueue.js @@ -10,6 +10,7 @@ // (/api/jobs/tick); both run the same code. import { getPoolConfig, sendToPool } from './dispatcher.js'; +import { FINALIZATION_RESERVE_MS, MIN_ATTEMPT_MS, remainingMs, routeDeadline } from './timeBudget.js'; export const TASK_ENDPOINTS = { CONVERT: '/api/v1/internal/process', @@ -164,7 +165,7 @@ export function spawnFromResult(item, body) { } /** Sends one claimed item to its pool and records the outcome. */ -export async function executeItem(db, item, { env = process.env, secret, fetchImpl } = {}) { +export async function executeItem(db, item, { env = process.env, secret, fetchImpl, deadlineMs = routeDeadline() } = {}) { const urls = getPoolConfig(env)[item.pool] || []; const started = Date.now(); const body = requestBody(item); @@ -174,6 +175,7 @@ export async function executeItem(db, item, { env = process.env, secret, fetchIm jobId: item.work_item_id, // spreads a job's pieces across both instances path: TASK_ENDPOINTS[item.task_type], secret, + deadlineMs, makeInit: () => ({ method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) }), ...(fetchImpl ? { fetchImpl } : {}), }); @@ -221,11 +223,12 @@ export async function executeItem(db, item, { env = process.env, secret, fetchIm * browser advancing its own upload); without it, any job (cron sweep). * Items of all pools run concurrently; the call returns when they finish. */ -export async function runTick(db, { jobId = null, env = process.env, secret, fetchImpl, limit = ITEMS_PER_TICK } = {}) { +export async function runTick(db, { jobId = null, env = process.env, secret, fetchImpl, limit = ITEMS_PER_TICK, deadlineMs = routeDeadline() } = {}) { await db.rpc('requeue_expired_work_items', {}); const capacity = poolCapacity(env); const claimed = []; for (const pool of POOLS) { + if (remainingMs(deadlineMs) < FINALIZATION_RESERVE_MS + MIN_ATTEMPT_MS) break; if (!capacity[pool]) continue; const rows = await db.rpc('claim_work_items', { p_pool: pool, @@ -239,7 +242,7 @@ export async function runTick(db, { jobId = null, env = process.env, secret, fet // expires and requeue_expired_work_items() retries it. const results = await Promise.all( claimed.map((item) => - executeItem(db, item, { env, secret, fetchImpl }).catch((err) => ({ + executeItem(db, item, { env, secret, fetchImpl, deadlineMs }).catch((err) => ({ item: item.work_item_id, outcome: 'ERROR', error: err.message, diff --git a/frontend/lib/server/jobService.js b/frontend/lib/server/jobService.js index 49bb56b..f2437e6 100644 --- a/frontend/lib/server/jobService.js +++ b/frontend/lib/server/jobService.js @@ -14,6 +14,7 @@ import { maxFileSizeFor, OCR_EXTENSIONS, poolForExtension, sizeLimitMessage, splitFileName } from '../formats.js'; import { getPoolConfig } from './dispatcher.js'; import { runTick } from './jobQueue.js'; +import { routeDeadline } from './timeBudget.js'; export const DOWNLOAD_URL_TTL_S = 600; @@ -101,7 +102,7 @@ async function loadJob(db, jobId) { return job; } -export async function startJob(db, jobId, { secret, env = process.env, fetchImpl, tick = true } = {}) { +export async function startJob(db, jobId, { secret, env = process.env, fetchImpl, tick = true, deadlineMs = routeDeadline() } = {}) { const job = await loadJob(db, jobId); if (TERMINAL.has(job.status) || job.status === 'CANCEL_REQUESTED') { return { status: 200, body: await jobStatus(db, jobId) }; @@ -118,15 +119,15 @@ export async function startJob(db, jobId, { secret, env = process.env, fetchImpl } await db.rpc('start_uploaded_job', { p_job_id: jobId, p_bucket: db.bucket, p_verified_size: size }); } - if (tick) await runTick(db, { jobId, env, secret, fetchImpl }); + if (tick) await runTick(db, { jobId, env, secret, fetchImpl, deadlineMs }); const body = await jobStatus(db, jobId); return { status: body.status === 'FAILED' ? statusForFailure(body.error?.code) : 200, body }; } -export async function advanceJob(db, jobId, { secret, env = process.env, fetchImpl } = {}) { +export async function advanceJob(db, jobId, { secret, env = process.env, fetchImpl, deadlineMs = routeDeadline() } = {}) { const job = await loadJob(db, jobId); if (!TERMINAL.has(job.status) && job.status !== 'UPLOADING') { - await runTick(db, { jobId, env, secret, fetchImpl }); + await runTick(db, { jobId, env, secret, fetchImpl, deadlineMs }); } return jobStatus(db, jobId); } diff --git a/frontend/lib/server/supabaseRest.js b/frontend/lib/server/supabaseRest.js index 2ea7b3c..a7635bb 100644 --- a/frontend/lib/server/supabaseRest.js +++ b/frontend/lib/server/supabaseRest.js @@ -4,6 +4,8 @@ // // Storage REST shapes follow @supabase/storage-js. +import { boundedTimeout } from './timeBudget.js'; + export function supabaseConfig(env = process.env) { const url = (env.SUPABASE_URL || '').replace(/\/+$/, ''); const key = env.SUPABASE_SERVICE_ROLE_KEY || ''; @@ -21,7 +23,7 @@ export class SupabaseError extends Error { const encodePath = (path) => path.split('/').map(encodeURIComponent).join('/'); -export function createSupabase(config, fetchImpl = fetch) { +export function createSupabase(config, fetchImpl = fetch, { deadlineMs } = {}) { const headers = { apikey: config.key, Authorization: `Bearer ${config.key}`, @@ -32,7 +34,7 @@ export function createSupabase(config, fetchImpl = fetch) { ...init, headers: { ...headers, ...(init.headers || {}) }, cache: 'no-store', - signal: AbortSignal.timeout(30_000), + signal: AbortSignal.timeout(boundedTimeout(deadlineMs, 30_000)), }); const text = await res.text(); let data = null; diff --git a/frontend/lib/server/timeBudget.js b/frontend/lib/server/timeBudget.js new file mode 100644 index 0000000..aea4b2c --- /dev/null +++ b/frontend/lib/server/timeBudget.js @@ -0,0 +1,23 @@ +// Per-route budgets, not a new lifetime limit on durable multi-stage jobs. +export const ROUTE_BUDGET_MS = 300_000; +export const FINALIZATION_RESERVE_MS = 60_000; +export const MIN_ATTEMPT_MS = 13_000; // backend save reserve + response margin + useful work +export const DEADLINE_HEADER = 'X-LLMiser-Deadline-Ms'; +export const DEADLINE_EXCEEDED_HEADER = 'X-LLMiser-Deadline-Exceeded'; +export const TIMEOUT_DETAIL = 'Conversion timed out. Try a smaller file or retry later.'; + +export class DeadlineExceededError extends Error { + constructor() { super(TIMEOUT_DETAIL); this.name = 'DeadlineExceededError'; this.status = 504; } +} + +export const routeDeadline = (now = Date.now()) => now + ROUTE_BUDGET_MS; + +export function remainingMs(deadlineMs, now = Date.now()) { + return Math.max(0, Math.floor(deadlineMs - now)); +} + +export function boundedTimeout(deadlineMs, ceilingMs, now = Date.now()) { + const remaining = deadlineMs === undefined ? ceilingMs : remainingMs(deadlineMs, now); + if (remaining < 1) throw new DeadlineExceededError(); + return Math.min(ceilingMs, remaining); +} diff --git a/frontend/test/timeBudget.test.mjs b/frontend/test/timeBudget.test.mjs new file mode 100644 index 0000000..b29903e --- /dev/null +++ b/frontend/test/timeBudget.test.mjs @@ -0,0 +1,92 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { sendToPool } from '../lib/server/dispatcher.js'; +import { executeItem, runTick } from '../lib/server/jobQueue.js'; +import { createSupabase } from '../lib/server/supabaseRest.js'; +import { DEADLINE_HEADER, DEADLINE_EXCEEDED_HEADER, DeadlineExceededError, boundedTimeout } from '../lib/server/timeBudget.js'; + +const response = (status, body = {}, headers = {}) => new Response(JSON.stringify(body), {status, headers}); +const request = (extra) => sendToPool({pool:'ocr', urls:['https://one.test','https://two.test'], jobId:'job', path:'/convert', secret:'s', makeInit:()=>({body:'generated'}), ...extra}); + +test('a slow transient failure leaves only the shared budget for the peer', async () => { + const origin = 1_800_000_000_000; + let clock = origin; + const deadlines = []; + const result = await request({deadlineMs:origin+300_000, now:()=>clock, fetchImpl:async (_, init)=>{ + deadlines.push(Number(init.headers[DEADLINE_HEADER])); + assert.equal(init.headers['X-Internal-Secret'],'s'); + if (deadlines.length === 1) {clock += 140_000; return response(503);} + return response(200, {complete:true}); + }}); + assert.equal(result.status,200); + assert.deepEqual(deadlines,[origin+148_000,origin+238_000]); + assert.equal(result.attempts,2); +}); + +test('exhausted budget skips the peer and preserves finalization time', async () => { + let clock = 1_800_000_000_000; + const deadlineMs = clock+300_000; + let calls = 0; + const result = await request({deadlineMs, now:()=>clock, fetchImpl:async()=>{calls++;clock+=230_000;return response(503);}}); + assert.equal(result.status,504); + assert.equal(result.attempts,1); + assert.equal(calls,1); +}); + +test('an expired route never constructs or sends another body', async () => { + const result = await request({deadlineMs:Date.now()+60_000, makeInit:()=>assert.fail('body constructed'), fetchImpl:()=>assert.fail('fetch called')}); + assert.equal(result.status,504); + assert.equal(result.attempts,0); +}); + +test('explicit backend deadline and response-body timeout do not retry', async () => { + for (const kind of ['backend','body']) { + let calls = 0; + const result = await request({fetchImpl:async()=>{ + calls++; + if (kind === 'backend') return response(504,{}, {[DEADLINE_EXCEEDED_HEADER]:'1'}); + return {status:200,ok:true,json:async()=>{throw new DOMException('Timed out','TimeoutError');}}; + }}); + assert.equal(result.status,504); + assert.equal(calls,1); + } +}); + +test('legacy backend 504 still permits the existing peer failover', async () => { + let calls = 0; + const result = await request({fetchImpl:async()=>response(++calls===1 ? 504 : 200)}); + assert.equal(result.status,200); + assert.equal(calls,2); +}); + +test('response decoding consuming the budget cannot report success', async () => { + let clock = 1_800_000_000_000; + const deadlineMs = clock+300_000; + const result = await request({deadlineMs, now:()=>clock,fetchImpl:async()=>({status:200,ok:true,json:async()=>{clock+=160_000;return {complete:true};}})}); + assert.equal(result.status,504); + assert.equal(result.attempts,1); +}); + +test('database calls cannot start after the outer route budget expires', async () => { + assert.equal(boundedTimeout(105,30,100),5); + const db = createSupabase({url:'https://db.test',key:'test',bucket:'test'},()=>assert.fail('expired REST call started'),{deadlineMs:Date.now()-1}); + await assert.rejects(db.rpc('finish_work_item',{}),DeadlineExceededError); +}); + +test('tick stops claiming new items when only finalization time remains', async () => { + const names = []; + const db = {rpc:async name=>{names.push(name);return [];}}; + const result = await runTick(db,{deadlineMs:Date.now()+60_000,env:{NORMAL_BACKEND_URLS:'https://one.test'}}); + assert.equal(result.claimed,0); + assert.deepEqual(names,['requeue_expired_work_items']); +}); + +test('an already claimed item records RETRY rather than false success on exhaustion', async () => { + const recorded = []; + const db = {bucket:'test',rpc:async(name,args)=>{recorded.push({name,args});return 'QUEUED';}}; + const item = {job_id:'job',work_item_id:'item',attempt_count:1,pool:'normal',task_type:'CONVERT',payload:{}}; + const result = await executeItem(db,item,{deadlineMs:Date.now()+60_000,env:{NORMAL_BACKEND_URLS:'https://one.test'},fetchImpl:()=>assert.fail('expired item dispatched')}); + assert.equal(result.outcome,'RETRY'); + assert.equal(recorded[0].args.p_error_code,'HTTP_504'); + assert.equal(recorded[0].args.p_result,null); +});