From 18c0c84367e70014a3e6ef9140d292370b958c1c Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Tue, 26 May 2026 16:58:52 +0800 Subject: [PATCH] up --- reme4/components/job/background_job.py | 37 +++++++++++++++++++++----- reme4/components/job/base_job.py | 27 ++++++++++++------- reme4/components/job/stream_job.py | 2 +- 3 files changed, 48 insertions(+), 18 deletions(-) diff --git a/reme4/components/job/background_job.py b/reme4/components/job/background_job.py index a0a4efa3..12251122 100644 --- a/reme4/components/job/background_job.py +++ b/reme4/components/job/background_job.py @@ -1,7 +1,9 @@ """Long-running background job with optional supervisor.""" import asyncio +import contextlib import random +import time from .base_job import BaseJob from ..component_registry import R @@ -18,6 +20,12 @@ class BackgroundJob(BaseJob): exponential backoff (backoff_base * 2**attempt, capped at backoff_cap) plus ±50% jitter. __call__ must NOT swallow exceptions, otherwise the supervisor cannot trigger a restart. + + On close, the stop_event is set and the task is given up to + ``close_timeout`` seconds to exit gracefully; after that it is cancelled. + If a single run survives at least ``attempt_reset_after`` seconds before + crashing, the backoff attempt counter resets — so a long-stable job that + eventually crashes restarts quickly rather than at the capped delay. """ def __init__( @@ -25,13 +33,17 @@ class BackgroundJob(BaseJob): supervisor: bool = True, backoff_base: float = 1.0, backoff_cap: float = 60.0, + close_timeout: float = 5.0, + attempt_reset_after: float = 60.0, **kwargs, ): super().__init__(**kwargs) self.supervisor: bool = supervisor self.backoff_base: float = backoff_base self.backoff_cap: float = backoff_cap - self._stop_event: asyncio.Event = asyncio.Event() + self.close_timeout: float = close_timeout + self.attempt_reset_after: float = attempt_reset_after + self._stop_event: asyncio.Event | None = None self._task: asyncio.Task | None = None async def _start(self) -> None: @@ -40,25 +52,35 @@ class BackgroundJob(BaseJob): self._task = asyncio.create_task(self._run_with_supervisor()) async def _close(self) -> None: - self._stop_event.set() + if self._stop_event is not None: + self._stop_event.set() if self._task is not None: try: - await self._task + await asyncio.wait_for(asyncio.shield(self._task), timeout=self.close_timeout) + except asyncio.TimeoutError: + self._task.cancel() + with contextlib.suppress(BaseException): + await self._task except Exception: self.logger.exception(f"Background task '{self.name}' raised during close") self._task = None await super()._close() async def _run_with_supervisor(self) -> None: + assert self._stop_event is not None attempt = 0 while not self._stop_event.is_set(): + started_at = time.monotonic() try: await self() return except Exception as e: if not self.supervisor: raise - delay = min(self.backoff_base * (2**attempt), self.backoff_cap) * (0.5 + random.random()) + if time.monotonic() - started_at >= self.attempt_reset_after: + attempt = 0 + capped = min(self.backoff_base * (2**attempt), self.backoff_cap) + delay = min(capped * (0.5 + random.random()), self.backoff_cap) self.logger.exception(f"job body crashed, restart in {delay:.2f}s error={e}") attempt += 1 try: @@ -67,8 +89,9 @@ class BackgroundJob(BaseJob): pass async def __call__(self, **kwargs) -> Response: - """Default body: run step_components in order; errors propagate to supervisor.""" - context = RuntimeContext(stop_event=self._stop_event, **self.kwargs) - for step in self.step_components: + """Default body: run steps in order; errors propagate to supervisor.""" + merged = {**self.kwargs, **kwargs} + context = RuntimeContext(stop_event=self._stop_event, **merged) + for step in self._build_steps(): await step(context) return context.response diff --git a/reme4/components/job/base_job.py b/reme4/components/job/base_job.py index 4928f2d0..0d2432cf 100644 --- a/reme4/components/job/base_job.py +++ b/reme4/components/job/base_job.py @@ -1,11 +1,16 @@ """Base job component for sequential step execution.""" +from typing import TYPE_CHECKING + from ..base_component import BaseComponent from ..component_registry import R from ..runtime_context import RuntimeContext from ...enumeration import ComponentEnum from ...schema import ComponentConfig, Response +if TYPE_CHECKING: + from ...steps import BaseStep + @R.register("base") class BaseJob(BaseComponent): @@ -24,14 +29,12 @@ class BaseJob(BaseComponent): self.description = description self.parameters = parameters or {} self.step_configs = steps or [] - - from ...steps import BaseStep - - self.step_components: list[BaseStep] = [] + self.step_specs: list[tuple[type["BaseStep"], dict]] = [] async def _start(self) -> None: - """Resolve step configs into instantiated step components.""" - assert self.app_context is not None, "app_context must be provided" + """Resolve step configs into (cls, params) pairs; defer instantiation to __call__.""" + if self.app_context is None: + raise RuntimeError(f"app_context must be provided for job '{self.name}'") for raw in self.step_configs: config = raw if isinstance(raw, ComponentConfig) else ComponentConfig(**raw) if not config.backend: @@ -41,17 +44,21 @@ class BaseJob(BaseComponent): raise ValueError(f"Unregistered backend '{config.backend}' of type '{ComponentEnum.STEP}'") params = config.model_dump() params["app_context"] = self.app_context - self.step_components.append(step_cls(**params)) + self.step_specs.append((step_cls, params)) async def _close(self) -> None: - """Release all step components.""" - self.step_components.clear() + """Release all step specs.""" + self.step_specs.clear() + + def _build_steps(self) -> list["BaseStep"]: + """Instantiate fresh step instances from stored specs.""" + return [step_cls(**dict(params)) for step_cls, params in self.step_specs] async def __call__(self, **kwargs) -> Response: """Execute all steps in order and return the final response.""" context = RuntimeContext(**kwargs) try: - for step in self.step_components: + for step in self._build_steps(): await step(context) except Exception as e: self.logger.exception(f"Failed to execute job: {e}") diff --git a/reme4/components/job/stream_job.py b/reme4/components/job/stream_job.py index 89f1fab1..193bf5e2 100644 --- a/reme4/components/job/stream_job.py +++ b/reme4/components/job/stream_job.py @@ -14,7 +14,7 @@ class StreamJob(BaseJob): """Execute steps and stream output; errors are sent as ERROR chunks.""" context = RuntimeContext(**kwargs) try: - for step in self.step_components: + for step in self._build_steps(): await step(context) except Exception as e: await context.add_stream_string(str(e), ChunkEnum.ERROR)