diff --git a/app/sep/apps/framework/spec.py b/app/sep/apps/framework/spec.py
index 2d369bd35..6d5655dbe 100644
--- a/app/sep/apps/framework/spec.py
+++ b/app/sep/apps/framework/spec.py
@@ -69,8 +69,8 @@
from app.sep.deps import get_created_entity, InventoryAPI
from app.sep.inventory import CreatedEntity, CreatedService
from app.sep.models import SyncInventoryEntityTypeEnum
+from app.tasks.execution.executors.nomad.constants import RUN_SCRIPT_OUTPUT_FILES_PATH
from app.tasks.models import (
- RUN_SCRIPT_OUTPUT_FILES_PATH,
TaskBackendEnum,
TaskWrite,
)
diff --git a/app/tasks/crud.py b/app/tasks/crud.py
index ba4f9f916..a5c1f14f0 100644
--- a/app/tasks/crud.py
+++ b/app/tasks/crud.py
@@ -305,24 +305,24 @@ class TaskHistoryManager(BaseSQLModelManager):
Model = TaskHistory
@classmethod
- async def get_log_allocation_epoch(
+ async def get_log_producer_epoch(
cls,
session: AsyncSession,
task_history_id: int,
*,
for_update: bool = False,
) -> int:
- """Return the task-level log allocation-epoch high-water mark.
+ """Return the task-level log producer-epoch high-water mark.
The log writer consults this on the first-insert path — before any
per-stream ``TaskHistoryLogState`` row exists — to discard writes from a
- superseded allocation. A missing row yields the ``0`` sentinel so the
+ superseded producer. A missing row yields the ``0`` sentinel so the
caller trusts the write.
With ``for_update`` the ``TaskHistory`` row is locked (``SELECT ... FOR
UPDATE``) and the lock is held until the caller's transaction ends. This
serialises the first-insert discard decision against
- :meth:`bump_log_allocation_epoch` (stamped during a frontier reset) so a
+ :meth:`bump_log_producer_epoch` (stamped during a frontier reset) so a
reset cannot commit a newer epoch in the window between the guard read
and the row insert. Both the first-insert writer and the frontier reset
acquire this same row first, giving a consistent lock order. On SQLite
@@ -333,10 +333,10 @@ async def get_log_allocation_epoch(
:param task_history_id: The ``TaskHistory`` identifier.
:param for_update: Whether to lock the row for the duration of the
transaction.
- :return: The stored ``log_allocation_epoch``, or ``0`` when the row is
+ :return: The stored ``log_producer_epoch``, or ``0`` when the row is
absent.
"""
- query = select(TaskHistory.log_allocation_epoch).where(
+ query = select(TaskHistory.log_producer_epoch).where(
col(TaskHistory.id) == task_history_id
)
if for_update:
@@ -346,14 +346,14 @@ async def get_log_allocation_epoch(
return epoch if epoch is not None else 0
@classmethod
- async def bump_log_allocation_epoch(
+ async def bump_log_producer_epoch(
cls,
session: AsyncSession,
task_history_id: int,
*,
- new_allocation_epoch: int,
+ new_producer_epoch: int,
) -> None:
- """Advance the task-level allocation-epoch high-water mark monotonically.
+ """Advance the task-level producer-epoch high-water mark monotonically.
Stamped wherever the log frontier is reset. The ``< new`` guard makes the
update monotonic: an out-of-order or stale reset carrying a smaller epoch
@@ -364,16 +364,16 @@ async def bump_log_allocation_epoch(
:param session: The SQLAlchemy asynchronous session to use for query
execution.
:param task_history_id: The ``TaskHistory`` identifier.
- :param new_allocation_epoch: The ``CreateIndex`` of the allocation the
- frontier is being reset onto.
+ :param new_producer_epoch: The producer epoch the frontier is being
+ reset onto (for example a Nomad allocation ``CreateIndex``).
"""
stmt = (
update(TaskHistory)
.where(
col(TaskHistory.id) == task_history_id,
- col(TaskHistory.log_allocation_epoch) < new_allocation_epoch,
+ col(TaskHistory.log_producer_epoch) < new_producer_epoch,
)
- .values(log_allocation_epoch=new_allocation_epoch)
+ .values(log_producer_epoch=new_producer_epoch)
)
await session.exec(stmt)
@@ -1172,8 +1172,8 @@ def build_default(
stream=stream,
persisted_offset=0,
producer_offset=0,
- nomad_offset=0,
- allocation_epoch=0,
+ producer_fetch_offset=0,
+ producer_epoch=0,
staging=b"",
staging_updated_at=utc_now(),
version=0,
@@ -1205,33 +1205,33 @@ async def reset_allocation_frontier(
session: AsyncSession,
task_history_id: int,
*,
- new_allocation_epoch: int,
+ new_producer_epoch: int,
) -> None:
"""Reset both cursors to zero and stamp the new epoch for every stream.
- Called when Nomad reschedules a task to a follow-up allocation: the
- new allocation's log file starts at byte 0, so both allocation-relative
- cursors (``producer_offset`` and ``nomad_offset``) must be cleared in
- the database before the writer dedups or fetches against them, and
- ``allocation_epoch`` must be advanced to the new allocation's
- ``CreateIndex`` so stale-allocation writes are discarded by the write
- guard. Bumps ``version`` so concurrent writers re-read the row.
+ Called when the executor reschedules onto a follow-up producer (for
+ example a Nomad allocation): the new producer's log file starts at byte
+ 0, so both producer-relative cursors (``producer_offset`` and
+ ``producer_fetch_offset``) must be cleared in the database before the
+ writer dedups or fetches against them, and ``producer_epoch`` must be
+ advanced so stale-producer writes are discarded by the write guard.
+ Bumps ``version`` so concurrent writers re-read the row.
:param session: The SQLAlchemy asynchronous session.
:type session: AsyncSession
:param task_history_id: The ``TaskHistory`` identifier whose state
rows should be reset.
:type task_history_id: int
- :param new_allocation_epoch: The ``CreateIndex`` of the allocation the
- frontier is being reset onto.
+ :param new_producer_epoch: The producer epoch the frontier is being
+ reset onto (for example a Nomad allocation ``CreateIndex``).
"""
stmt = (
update(TaskHistoryLogState)
.where(col(TaskHistoryLogState.task_history_id) == task_history_id)
.values(
producer_offset=0,
- nomad_offset=0,
- allocation_epoch=new_allocation_epoch,
+ producer_fetch_offset=0,
+ producer_epoch=new_producer_epoch,
version=col(TaskHistoryLogState.version) + 1,
updated_at=utc_now(),
)
@@ -1248,8 +1248,8 @@ async def insert_row_idempotent(
stream: TaskLogType,
persisted_offset: int,
producer_offset: int,
- nomad_offset: int,
- allocation_epoch: int,
+ producer_fetch_offset: int,
+ producer_epoch: int,
staging: bytes,
version: int,
now: datetime,
@@ -1272,10 +1272,12 @@ async def insert_row_idempotent(
:param persisted_offset: The user-facing byte offset already persisted.
:type persisted_offset: int
:param producer_offset: The producer-relative byte offset already
- consumed from the current allocation.
+ consumed from the current producer epoch.
:type producer_offset: int
- :param nomad_offset: The raw Nomad-space fetch offset for the next read.
- :param allocation_epoch: The Nomad ``CreateIndex`` the cursors belong to.
+ :param producer_fetch_offset: The raw producer-space fetch offset for
+ the next read.
+ :param producer_epoch: The producer-generation stamp the cursors belong
+ to (for example a Nomad allocation ``CreateIndex``).
:param staging: Bytes pending flush to the chunk store.
:type staging: bytes
:param version: The initial optimistic-locking version counter.
@@ -1292,8 +1294,8 @@ async def insert_row_idempotent(
stream=stream,
persisted_offset=persisted_offset,
producer_offset=producer_offset,
- nomad_offset=nomad_offset,
- allocation_epoch=allocation_epoch,
+ producer_fetch_offset=producer_fetch_offset,
+ producer_epoch=producer_epoch,
staging=staging,
staging_updated_at=now,
version=version,
@@ -1314,8 +1316,8 @@ async def update_row_if_version(
new_version: int,
persisted_offset: int,
producer_offset: int,
- nomad_offset: int,
- allocation_epoch: int,
+ producer_fetch_offset: int,
+ producer_epoch: int,
staging: bytes,
now: datetime,
) -> bool:
@@ -1343,9 +1345,10 @@ async def update_row_if_version(
:type persisted_offset: int
:param producer_offset: The updated producer-relative offset.
:type producer_offset: int
- :param nomad_offset: The updated raw Nomad-space fetch offset.
- :param allocation_epoch: The updated Nomad ``CreateIndex`` the cursors
- belong to.
+ :param producer_fetch_offset: The updated raw producer-space fetch
+ offset.
+ :param producer_epoch: The updated producer-generation stamp the
+ cursors belong to (for example a Nomad allocation ``CreateIndex``).
:param staging: The updated staging bytes buffer.
:type staging: bytes
:param now: The update timestamp used for the audit columns.
@@ -1365,8 +1368,8 @@ async def update_row_if_version(
.values(
persisted_offset=persisted_offset,
producer_offset=producer_offset,
- nomad_offset=nomad_offset,
- allocation_epoch=allocation_epoch,
+ producer_fetch_offset=producer_fetch_offset,
+ producer_epoch=producer_epoch,
staging=staging,
staging_updated_at=now,
version=new_version,
diff --git a/app/tasks/db/seed.py b/app/tasks/db/seed.py
index 24e5357bb..b1162dd75 100644
--- a/app/tasks/db/seed.py
+++ b/app/tasks/db/seed.py
@@ -35,10 +35,12 @@
from app.tasks.crud import TaskManager
from app.tasks.db import get_async_session_maker
from app.tasks.db.engine import engine
-from app.tasks.models import (
+from app.tasks.execution.executors.nomad.constants import (
CHECK_NOMAD_CERT_EXPIRY_TASK_NAME,
- INVENTORY_SYNC_TASK_NAME,
RUN_SCRIPT_OUTPUT_FILES_PATH,
+)
+from app.tasks.models import (
+ INVENTORY_SYNC_TASK_NAME,
SYNC_RUNNING_TASKS_TASK_NAME,
SYSTEM_USER,
Task,
diff --git a/app/tasks/execution/executors/nomad/__init__.py b/app/tasks/execution/executors/nomad/__init__.py
index 003dba125..2ddf42bfc 100644
--- a/app/tasks/execution/executors/nomad/__init__.py
+++ b/app/tasks/execution/executors/nomad/__init__.py
@@ -13,4 +13,31 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
-from app.tasks.execution.executors.nomad.models import NomadExecutor
+"""Nomad task executor package.
+
+``NomadExecutor`` is resolved lazily via ``__getattr__`` so submodule imports
+do not deadlock against ``app.tasks.config`` (which imports ``NomadExecutor``
+back from this package).
+"""
+
+__all__ = ["NomadExecutor"]
+
+
+def __getattr__(name: str) -> object:
+ """Resolve ``NomadExecutor`` on first attribute access.
+
+ :param name: The attribute being read.
+ :return: The resolved attribute.
+ :raises AttributeError: If ``name`` is not exported by this package.
+ """
+ if name == "NomadExecutor":
+ from app.tasks.execution.executors.nomad.models import NomadExecutor
+
+ globals()[name] = NomadExecutor
+ return NomadExecutor
+ raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def __dir__() -> list[str]:
+ """Return attribute names for ``dir()``, including the lazy export."""
+ return sorted({*globals(), *__all__})
diff --git a/app/tasks/execution/executors/nomad/constants.py b/app/tasks/execution/executors/nomad/constants.py
new file mode 100644
index 000000000..446f79e21
--- /dev/null
+++ b/app/tasks/execution/executors/nomad/constants.py
@@ -0,0 +1,33 @@
+# Copyright (C) 2026 Percona LLC
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see .
+
+"""Define Nomad-owned constants for allocation layout and Nomad-only system tasks.
+
+Imports nothing from the rest of the tasks or sep packages so seed and framework
+specs can use these values without the Nomad executor import graph.
+"""
+
+#: Allocation-relative output-files directory of every job spec that pins its
+#: ``run-script`` task's ``work_dir`` to ``${NOMAD_TASK_DIR}/output_files``
+#: (``run-python``, ``exec-artifact``, ``exec-python-artifact``). It is the
+#: :attr:`~app.tasks.models.TaskBase.output_files_path` those specs run under,
+#: so a payload's working directory and the path SEP reads its files back from
+#: are the same place. ``run-command`` pins no ``work_dir`` and so has no
+#: output-files path. The ``run-script/local/`` prefix is the Nomad allocation
+#: layout (``run-script`` task name + ``${NOMAD_TASK_DIR}``).
+RUN_SCRIPT_OUTPUT_FILES_PATH = "run-script/local/output_files"
+
+#: Seeded name of the Nomad-only periodic task that checks TLS cert expiry.
+CHECK_NOMAD_CERT_EXPIRY_TASK_NAME = "tasks__check_nomad_cert_expiry"
diff --git a/app/tasks/execution/executors/nomad/models.py b/app/tasks/execution/executors/nomad/models.py
index ec153f684..2d5cf2232 100644
--- a/app/tasks/execution/executors/nomad/models.py
+++ b/app/tasks/execution/executors/nomad/models.py
@@ -59,7 +59,6 @@
utc_now,
)
from app.core.utils.pydantic import field_with_metadata
-from app.tasks import config as tasks_config
from app.tasks.anonymizer import anonymize_text
from app.tasks.anonymizer.entities import PIIEntity
from app.tasks.crud import TaskHistoryLogStateManager, TaskHistoryManager
@@ -499,6 +498,10 @@ def dispatch_job(
parameterized_job.get("MetaRequired") or []
)
if "staleness_threshold_seconds" in declared_meta:
+ # Lazy import keeps app.tasks.config out of the nomad.models import
+ # chain (config imports NomadExecutor back from this package).
+ from app.tasks import config as tasks_config
+
filtered_meta["staleness_threshold_seconds"] = str(
tasks_config.tasks_settings.STALENESS_THRESHOLD_SECONDS
)
@@ -768,7 +771,7 @@ def _fetch_step_log_delta(
:param anonymize_entities: PII entities to anonymize for ``run-script``
and ``step1`` content. When ``None``, no anonymization is performed.
:type anonymize_entities: set[PIIEntity] | None
- :return: ``(delta_text, new_nomad_offset, new_producer_offset)`` —
+ :return: ``(delta_text, new_producer_fetch_offset, new_producer_offset)`` —
the anonymized bytes fetched this cycle, the advanced Nomad-space
offset for the next fetch, and the advanced producer-space offset
that the writer should persist.
@@ -790,20 +793,20 @@ def _fetch_step_log_delta(
)
return "", nomad_start_offset, producer_start_offset
pieces = []
- nomad_offset = nomad_start_offset
+ producer_fetch_offset = nomad_start_offset
for raw_log_data_item in (
"{" + item for item in raw_log_data.split("{") if item
):
log_data = json.loads(raw_log_data_item)
if raw_msg := log_data.get("Data"):
- nomad_offset = log_data.get("Offset", nomad_offset)
+ producer_fetch_offset = log_data.get("Offset", producer_fetch_offset)
msg = b64decode_str(raw_msg)
if step in ("run-script", "step1") and anonymize_entities:
msg = anonymize_text(msg, anonymize_entities)
pieces.append(msg)
delta = "".join(pieces)
producer_offset = producer_start_offset + len(delta.encode("utf-8"))
- return delta, nomad_offset, producer_offset
+ return delta, producer_fetch_offset, producer_offset
def get_logs_for_allocation(
self,
@@ -854,15 +857,17 @@ def get_logs_for_allocation(
producer_offset_key = f"{log_type}_producer_offset"
nomad_start_offset = task_logs[step].get(last_offset_key) or 0
producer_start_offset = task_logs[step].get(producer_offset_key) or 0
- delta, new_nomad_offset, new_producer_offset = self._fetch_step_log_delta(
- alloc_id,
- step,
- log_type,
- nomad_start_offset,
- producer_start_offset,
- anonymize_entities,
+ delta, new_producer_fetch_offset, new_producer_offset = (
+ self._fetch_step_log_delta(
+ alloc_id,
+ step,
+ log_type,
+ nomad_start_offset,
+ producer_start_offset,
+ anonymize_entities,
+ )
)
- task_logs[step][last_offset_key] = new_nomad_offset
+ task_logs[step][last_offset_key] = new_producer_fetch_offset
task_logs[step][producer_offset_key] = new_producer_offset
task_logs[step][log_type] = delta
return task_logs
@@ -999,7 +1004,7 @@ async def _persist_nomad_task_logs(
) -> None:
"""Persist this sync cycle's delta logs into the chunk store.
- Resets the fetch frontier (both cursors zeroed, ``allocation_epoch``
+ Resets the fetch frontier (both cursors zeroed, ``producer_epoch``
stamped to the new allocation's ``CreateIndex``) for every stream when
Nomad reschedules to a new allocation so the fetcher reads the new
allocation from the start; the epoch is threaded into every write so a
@@ -1025,7 +1030,7 @@ async def _persist_nomad_task_logs(
if previous_allocation_id is not None and previous_allocation_id != alloc_id:
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- writer_session, queue_item.id, new_allocation_epoch=alloc_epoch
+ writer_session, queue_item.id, new_producer_epoch=alloc_epoch
)
initial_offsets = await self._build_initial_log_offsets(
@@ -1126,7 +1131,7 @@ async def _build_initial_log_offsets(
Seeds each stream's next-fetch cursor from its durable
``TaskHistoryLogState`` row, but only when the row belongs to the
- current allocation — its ``allocation_epoch`` matches ``current_epoch``
+ current allocation — its ``producer_epoch`` matches ``current_epoch``
or is the ``0`` legacy/unknown sentinel that trusts the migration
backfill. A row stamped to a known *different* allocation holds a cursor
in that allocation's byte space, so it is skipped and the stream
@@ -1147,10 +1152,10 @@ async def _build_initial_log_offsets(
)
initial_offsets = defaultdict(dict)
for row in state_rows:
- if row.allocation_epoch not in (0, current_epoch):
+ if row.producer_epoch not in (0, current_epoch):
continue
initial_offsets[row.source][f"{row.stream.value}_last_offset"] = (
- row.nomad_offset
+ row.producer_fetch_offset
)
initial_offsets[row.source][f"{row.stream.value}_producer_offset"] = (
row.producer_offset
@@ -1187,7 +1192,7 @@ async def _write_nomad_deltas(
delta_text = payload.get(log_type) or ""
if not delta_text and not force_flush:
continue
- nomad_offset = payload.get(f"{log_type.value}_last_offset", 0)
+ producer_fetch_offset = payload.get(f"{log_type.value}_last_offset", 0)
producer_offset = payload.get(f"{log_type.value}_producer_offset", 0)
await TaskHistoryLogWriter.append(
writer_session,
@@ -1196,8 +1201,8 @@ async def _write_nomad_deltas(
stream=log_type,
new_bytes=delta_text.encode("utf-8"),
producer_offset_after=producer_offset,
- nomad_offset_after=nomad_offset,
- allocation_epoch=alloc_epoch,
+ producer_fetch_offset_after=producer_fetch_offset,
+ producer_epoch=alloc_epoch,
force_flush=force_flush,
)
diff --git a/app/tasks/logs/log_writer.py b/app/tasks/logs/log_writer.py
index 58413a6f3..c5f01b4a2 100644
--- a/app/tasks/logs/log_writer.py
+++ b/app/tasks/logs/log_writer.py
@@ -21,7 +21,6 @@
from sqlmodel.ext.asyncio.session import AsyncSession
from app.core.utils.date_time import utc_now
-from app.tasks import config as tasks_config
from app.tasks.crud import (
TaskHistoryLogManager,
TaskHistoryLogStateManager,
@@ -70,8 +69,8 @@ async def append(
new_bytes: bytes,
force_flush: bool = False,
producer_offset_after: int | None = None,
- nomad_offset_after: int | None = None,
- allocation_epoch: int | None = None,
+ producer_fetch_offset_after: int | None = None,
+ producer_epoch: int | None = None,
) -> None:
"""Persist ``new_bytes`` for the given ``(task_history_id, source, stream)``.
@@ -94,19 +93,20 @@ async def append(
end of ``new_bytes``. When provided, the state row's
``producer_offset`` is advanced atomically with the flush.
:type producer_offset_after: int | None
- :param nomad_offset_after: The raw Nomad-space fetch offset for the next
- read. When provided, the row's ``nomad_offset`` is advanced
- atomically with the flush; when ``None`` the existing value is
- preserved so non-Nomad callers do not disturb it.
- :param allocation_epoch: The Nomad allocation ``CreateIndex`` the bytes
- belong to. When it is *older* than the current allocation epoch the
- write is discarded — the bytes come from an allocation the frontier
- has already moved past (a sync that overlapped a reschedule) and
- appending them would corrupt the stream. For an existing row the
- comparison is against that row's per-stream epoch; on the
- first-insert path (no row yet) it is against the task-level
- high-water mark stamped at the last frontier reset.
- ``None`` leaves the row's epoch untouched (non-Nomad callers).
+ :param producer_fetch_offset_after: The raw producer-space fetch offset
+ for the next read. When provided, the row's ``producer_fetch_offset``
+ is advanced atomically with the flush; when ``None`` the existing
+ value is preserved so callers that do not track a fetch cursor do
+ not disturb it.
+ :param producer_epoch: The producer-generation stamp the bytes belong
+ to (for example a Nomad allocation ``CreateIndex``). When it is
+ *older* than the current producer epoch the write is discarded —
+ the bytes come from a producer the frontier has already moved past
+ (a sync that overlapped a reschedule) and appending them would
+ corrupt the stream. For an existing row the comparison is against
+ that row's per-stream epoch; on the first-insert path (no row yet)
+ it is against the task-level high-water mark stamped at the last
+ frontier reset. ``None`` leaves the row's epoch untouched.
:raises LogWriterConflictError: If the optimistic-locking retries are
exhausted without converging on a successful update.
"""
@@ -120,18 +120,18 @@ async def append(
task_history_id, source, stream
)
- if allocation_epoch is not None:
+ if producer_epoch is not None:
# First insert has no per-stream row yet; guard against the
# task-level high-water mark, not the transient row's ``0``.
- guard_epoch = state.allocation_epoch
+ guard_epoch = state.producer_epoch
if is_new:
# Take the row lock so a concurrent frontier reset serialises
# instead of racing (first-insert TOCTOU).
- guard_epoch = await TaskHistoryManager.get_log_allocation_epoch(
+ guard_epoch = await TaskHistoryManager.get_log_producer_epoch(
session, task_history_id, for_update=True
)
- if allocation_epoch < guard_epoch:
- # Stale write from a superseded allocation; drop it, releasing
+ if producer_epoch < guard_epoch:
+ # Stale write from a superseded producer; drop it, releasing
# any first-insert lock so it doesn't pin the row.
await cls._release_first_insert_lock(session, is_new=is_new)
return
@@ -187,15 +187,13 @@ async def append(
if producer_offset_after is not None
else state.producer_offset
)
- new_nomad = (
- nomad_offset_after
- if nomad_offset_after is not None
- else state.nomad_offset
+ new_fetch = (
+ producer_fetch_offset_after
+ if producer_fetch_offset_after is not None
+ else state.producer_fetch_offset
)
- new_allocation_epoch = (
- allocation_epoch
- if allocation_epoch is not None
- else state.allocation_epoch
+ new_producer_epoch = (
+ producer_epoch if producer_epoch is not None else state.producer_epoch
)
new_version = state.version + 1
@@ -209,8 +207,8 @@ async def append(
old_version=state.version,
persisted_offset=persisted_offset,
producer_offset=new_producer,
- nomad_offset=new_nomad,
- allocation_epoch=new_allocation_epoch,
+ producer_fetch_offset=new_fetch,
+ producer_epoch=new_producer_epoch,
staging=staging,
now=now,
)
@@ -238,30 +236,30 @@ async def drain_and_reset_allocation_frontier(
session: AsyncSession,
task_history_id: int,
*,
- new_allocation_epoch: int,
+ new_producer_epoch: int,
) -> None:
"""Flush every stream's staging buffer and reset the fetch frontier.
- Called when Nomad reschedules a task to a follow-up allocation: the
- new allocation's log file starts at byte 0, so the allocation-relative
- cursors (``producer_offset`` and ``nomad_offset``) from the previous
- allocation must be cleared and ``allocation_epoch`` advanced to the new
- allocation's ``CreateIndex``, and the leftover staging bytes from the
- previous allocation must be emitted as their own chunk instead of being
- concatenated with the new allocation's bytes.
+ Called when the executor reschedules onto a follow-up producer (for
+ example a Nomad allocation): the new producer's log file starts at byte
+ 0, so the producer-relative cursors (``producer_offset`` and
+ ``producer_fetch_offset``) from the previous producer must be cleared
+ and ``producer_epoch`` advanced, and the leftover staging bytes from
+ the previous producer must be emitted as their own chunk instead of
+ being concatenated with the new producer's bytes.
:param session: The SQLAlchemy asynchronous session.
:type session: AsyncSession
:param task_history_id: The ``TaskHistory`` identifier whose state
rows should be drained and reset.
:type task_history_id: int
- :param new_allocation_epoch: The ``CreateIndex`` of the allocation the
- frontier is being reset onto.
+ :param new_producer_epoch: The producer epoch the frontier is being
+ reset onto (for example a Nomad allocation ``CreateIndex``).
"""
# Lock the TaskHistory row before touching any log/state rows so this
# reset and a concurrent first-insert append serialise in the same order
# (TaskHistory first), closing the first-insert TOCTOU.
- await TaskHistoryManager.get_log_allocation_epoch(
+ await TaskHistoryManager.get_log_producer_epoch(
session, task_history_id, for_update=True
)
rows = await TaskHistoryLogStateManager.list_for_task(session, task_history_id)
@@ -287,8 +285,8 @@ async def drain_and_reset_allocation_frontier(
new_version=row.version + 1,
persisted_offset=new_persisted,
producer_offset=row.producer_offset,
- nomad_offset=row.nomad_offset,
- allocation_epoch=row.allocation_epoch,
+ producer_fetch_offset=row.producer_fetch_offset,
+ producer_epoch=row.producer_epoch,
staging=b"",
now=now,
)
@@ -310,13 +308,13 @@ async def drain_and_reset_allocation_frontier(
},
)
await TaskHistoryLogStateManager.reset_allocation_frontier(
- session, task_history_id, new_allocation_epoch=new_allocation_epoch
+ session, task_history_id, new_producer_epoch=new_producer_epoch
)
# Stamp the task-level high-water mark in the same transaction as the
# per-stream reset so a first-insert guard (no per-stream row yet) has a
# current epoch to check against.
- await TaskHistoryManager.bump_log_allocation_epoch(
- session, task_history_id, new_allocation_epoch=new_allocation_epoch
+ await TaskHistoryManager.bump_log_producer_epoch(
+ session, task_history_id, new_producer_epoch=new_producer_epoch
)
await session.commit()
@@ -542,8 +540,8 @@ async def _persist_state(
old_version: int,
persisted_offset: int,
producer_offset: int,
- nomad_offset: int,
- allocation_epoch: int,
+ producer_fetch_offset: int,
+ producer_epoch: int,
staging: bytes,
now: datetime,
) -> bool:
@@ -569,8 +567,9 @@ async def _persist_state(
:type persisted_offset: int
:param producer_offset: The producer-relative offset to persist.
:type producer_offset: int
- :param nomad_offset: The raw Nomad-space fetch offset to persist.
- :param allocation_epoch: The Nomad ``CreateIndex`` to persist.
+ :param producer_fetch_offset: The raw producer-space fetch offset to
+ persist.
+ :param producer_epoch: The producer-generation stamp to persist.
:param staging: The remaining staging buffer to persist.
:type staging: bytes
:param now: The update timestamp.
@@ -586,8 +585,8 @@ async def _persist_state(
stream=stream,
persisted_offset=persisted_offset,
producer_offset=producer_offset,
- nomad_offset=nomad_offset,
- allocation_epoch=allocation_epoch,
+ producer_fetch_offset=producer_fetch_offset,
+ producer_epoch=producer_epoch,
staging=staging,
version=new_version,
now=now,
@@ -601,8 +600,8 @@ async def _persist_state(
new_version=new_version,
persisted_offset=persisted_offset,
producer_offset=producer_offset,
- nomad_offset=nomad_offset,
- allocation_epoch=allocation_epoch,
+ producer_fetch_offset=producer_fetch_offset,
+ producer_epoch=producer_epoch,
staging=staging,
now=now,
)
@@ -635,6 +634,10 @@ async def _evict_over_cap(
"""
if persisted_offset <= previous_persisted_offset:
return
+ # Lazy import keeps app.tasks.config out of the log_writer import chain
+ # (log_writer sits on the NomadExecutor ↔ config cycle).
+ from app.tasks import config as tasks_config
+
cap = tasks_config.tasks_settings.LOG_STREAM_CAP_BYTES
low_water = persisted_offset - cap
if low_water <= 0:
diff --git a/app/tasks/migrations/versions/2026_07_30_1300-c8e4a2b91f70_rename_log_cursor_columns_executor_neutral.py b/app/tasks/migrations/versions/2026_07_30_1300-c8e4a2b91f70_rename_log_cursor_columns_executor_neutral.py
new file mode 100644
index 000000000..5f15f21f5
--- /dev/null
+++ b/app/tasks/migrations/versions/2026_07_30_1300-c8e4a2b91f70_rename_log_cursor_columns_executor_neutral.py
@@ -0,0 +1,73 @@
+# Copyright (C) 2026 Percona LLC
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see .
+
+"""rename Nomad-named log cursor columns to executor-neutral names
+
+Revision ID: c8e4a2b91f70
+Revises: 27a11549ef43
+Create Date: 2026-07-30 13:00:00.000000
+
+Rename the three Nomad-vocabulary log-cursor columns in place (metadata-only;
+values preserved):
+
+* ``taskhistory_log_state.nomad_offset`` → ``producer_fetch_offset``
+* ``taskhistory_log_state.allocation_epoch`` → ``producer_epoch``
+* ``taskhistory.log_allocation_epoch`` → ``log_producer_epoch``
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+
+revision: str = "c8e4a2b91f70"
+down_revision: Union[str, None] = "27a11549ef43"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+ op.alter_column(
+ "taskhistory_log_state",
+ "nomad_offset",
+ new_column_name="producer_fetch_offset",
+ )
+ op.alter_column(
+ "taskhistory_log_state",
+ "allocation_epoch",
+ new_column_name="producer_epoch",
+ )
+ op.alter_column(
+ "taskhistory",
+ "log_allocation_epoch",
+ new_column_name="log_producer_epoch",
+ )
+
+
+def downgrade() -> None:
+ op.alter_column(
+ "taskhistory",
+ "log_producer_epoch",
+ new_column_name="log_allocation_epoch",
+ )
+ op.alter_column(
+ "taskhistory_log_state",
+ "producer_epoch",
+ new_column_name="allocation_epoch",
+ )
+ op.alter_column(
+ "taskhistory_log_state",
+ "producer_fetch_offset",
+ new_column_name="nomad_offset",
+ )
diff --git a/app/tasks/models.py b/app/tasks/models.py
index df8f484c9..c82ae8a51 100644
--- a/app/tasks/models.py
+++ b/app/tasks/models.py
@@ -66,14 +66,6 @@
SYSTEM_USER = "SYSTEM"
ANY_OWNER = "ANY"
-#: Allocation-relative output-files directory of every job spec that pins its
-#: ``run-script`` task's ``work_dir`` to ``${NOMAD_TASK_DIR}/output_files``
-#: (``run-python``, ``exec-artifact``, ``exec-python-artifact``). It is the
-#: :attr:`TaskBase.output_files_path` those specs run under, so a payload's
-#: working directory and the path SEP reads its files back from are the same
-#: place. ``run-command`` pins no ``work_dir`` and so has no output-files path.
-RUN_SCRIPT_OUTPUT_FILES_PATH = "run-script/local/output_files"
-
def _encode_anonymize_mask(v: Any) -> Any:
"""Encode the anonymize mask from a set of PII entities.
@@ -110,11 +102,13 @@ class ExecutionEvent(BaseModel):
:param timestamp: When the event occurred (UTC).
:type timestamp: UTCDatetime
- :param event_type: Executor-specific event category (e.g. Nomad task event type).
+ :param event_type: Executor-specific event category (for example a Nomad
+ task event type).
:type event_type: str
:param description: Human-readable message for the event (no step prefix).
:type description: str
- :param step: Optional executor task/step name (e.g. Nomad task within the group).
+ :param step: Optional executor task/step name (for example a Nomad task
+ within the group).
:type step: str | None
"""
@@ -698,12 +692,12 @@ class TaskHistory(TaskHistoryBase, BaseSQLModel, table=True):
:type task: Task
:param sync_in_progress_started_at: Timestamp lock for a sync currently in progress.
:type sync_in_progress_started_at: UTCDatetime | None
- :param log_allocation_epoch: Task-level high-water mark of the current Nomad
- allocation ``CreateIndex``, stamped whenever the log frontier is reset. The
- log writer consults it on the first-insert path (before any per-stream
- ``TaskHistoryLogState`` row exists) to discard writes from a superseded
- allocation. ``0`` is the legacy/unknown sentinel that is trusted
- unconditionally.
+ :param log_producer_epoch: Task-level high-water mark of the current
+ producer epoch (for example a Nomad allocation ``CreateIndex``), stamped
+ whenever the log frontier is reset. The log writer consults it on the
+ first-insert path (before any per-stream ``TaskHistoryLogState`` row
+ exists) to discard writes from a superseded producer. ``0`` is the
+ legacy/unknown sentinel that is trusted unconditionally.
:param executed_by: The user ID of the user who executed the task.
:type executed_by: str | None
"""
@@ -722,7 +716,7 @@ class TaskHistory(TaskHistoryBase, BaseSQLModel, table=True):
default=None,
sa_type=DateTimeWithTimezone,
)
- log_allocation_epoch: int = SQLField(
+ log_producer_epoch: int = SQLField(
sa_column=Column(
BigInteger,
nullable=False,
@@ -883,7 +877,8 @@ class TaskHistoryLogState(BaseSQLModel, table=True):
:param task_history_id: The ID of the ``TaskHistory`` this state row tracks.
:type task_history_id: int
- :param source: The execution step (Nomad task name) this state row tracks.
+ :param source: The execution step name this state row tracks (for example a
+ Nomad task name).
:type source: str
:param stream: The log stream (stdout or stderr) this state row tracks.
:type stream: TaskLogType
@@ -891,19 +886,21 @@ class TaskHistoryLogState(BaseSQLModel, table=True):
flushed into the chunk store.
:type persisted_offset: int
:param producer_offset: The producer-relative byte offset already consumed
- from the current allocation (Nomad-relative; resets on allocation
- switch). May diverge from ``persisted_offset`` after a Nomad
- followup-allocation switch.
+ from the current producer epoch (resets when the producer switches, for
+ example on a Nomad follow-up allocation). May diverge from
+ ``persisted_offset`` after such a switch.
:type producer_offset: int
- :param nomad_offset: The raw Nomad-space byte offset for the next log fetch
- (the ``offset=`` kwarg of the next ``stream_logs.stream`` call).
- Allocation-relative like ``producer_offset`` (resets on switch); kept
- durable here so a worker without the in-memory cursor resumes the fetch
- instead of re-reading the whole file from ``0``.
- :param allocation_epoch: The Nomad allocation ``CreateIndex`` (monotonic,
- creation-anchored) that ``nomad_offset``/``producer_offset`` belong to.
- ``0`` is the legacy/unknown sentinel (pre-migration rows and non-Nomad
- streams) that the seed and write guards trust unconditionally.
+ :param producer_fetch_offset: The raw producer-space byte offset for the
+ next log fetch (for example the ``offset=`` kwarg of a Nomad
+ ``stream_logs.stream`` call). Producer-relative like
+ ``producer_offset`` (resets on switch); kept durable here so a worker
+ without the in-memory cursor resumes the fetch instead of re-reading
+ the whole file from ``0``.
+ :param producer_epoch: Monotonic producer-generation stamp that
+ ``producer_fetch_offset`` / ``producer_offset`` belong to (for example
+ a Nomad allocation ``CreateIndex``). ``0`` is the legacy/unknown
+ sentinel (pre-migration rows and streams without a producer epoch)
+ that the seed and write guards trust unconditionally.
:param staging: Bytes pending flush to the chunk store.
:type staging: bytes
:param staging_updated_at: When ``staging`` was last modified; used to age
@@ -951,14 +948,14 @@ class TaskHistoryLogState(BaseSQLModel, table=True):
server_default="0",
),
)
- nomad_offset: int = SQLField(
+ producer_fetch_offset: int = SQLField(
sa_column=Column(
BigInteger,
nullable=False,
server_default="0",
),
)
- allocation_epoch: int = SQLField(
+ producer_epoch: int = SQLField(
sa_column=Column(
BigInteger,
nullable=False,
@@ -992,13 +989,16 @@ class TaskHistoryLogState(BaseSQLModel, table=True):
INVENTORY_SYNC_TASK_NAME = "inventory-sync"
SYNC_RUNNING_TASKS_TASK_NAME = "tasks__sync_running_tasks"
-CHECK_NOMAD_CERT_EXPIRY_TASK_NAME = "tasks__check_nomad_cert_expiry"
+#: Maintenance / system task names excluded from user-facing task lists.
+#: The cert-expiry member is a literal matching
+#: :data:`~app.tasks.execution.executors.nomad.constants.CHECK_NOMAD_CERT_EXPIRY_TASK_NAME`
+#: so this module does not import the Nomad executor package.
INTERNAL_TASK_NAMES: frozenset[str] = frozenset(
{
INVENTORY_SYNC_TASK_NAME,
SYNC_RUNNING_TASKS_TASK_NAME,
- CHECK_NOMAD_CERT_EXPIRY_TASK_NAME,
+ "tasks__check_nomad_cert_expiry",
}
)
diff --git a/changelog.d/SEP-1630.breaking.md b/changelog.d/SEP-1630.breaking.md
new file mode 100644
index 000000000..defb22c47
--- /dev/null
+++ b/changelog.d/SEP-1630.breaking.md
@@ -0,0 +1 @@
+TaskHistory responses from the tasks service rename ``log_allocation_epoch`` to ``log_producer_epoch``. External consumers of ``POST /history/`` (and any other endpoint that returns ``TaskHistory``) must read the new field name; the value and semantics are unchanged.
diff --git a/frontend/packages/api/specs/sep.json b/frontend/packages/api/specs/sep.json
index 06fa64190..f5e61e8e5 100644
--- a/frontend/packages/api/specs/sep.json
+++ b/frontend/packages/api/specs/sep.json
@@ -314,7 +314,7 @@
"type": "object"
},
"ExecutionEvent": {
- "description": "A single lifecycle event from a task executor (executor-agnostic shape).\n\n:param timestamp: When the event occurred (UTC).\n:type timestamp: UTCDatetime\n:param event_type: Executor-specific event category (e.g. Nomad task event type).\n:type event_type: str\n:param description: Human-readable message for the event (no step prefix).\n:type description: str\n:param step: Optional executor task/step name (e.g. Nomad task within the group).\n:type step: str | None",
+ "description": "A single lifecycle event from a task executor (executor-agnostic shape).\n\n:param timestamp: When the event occurred (UTC).\n:type timestamp: UTCDatetime\n:param event_type: Executor-specific event category (for example a Nomad\n task event type).\n:type event_type: str\n:param description: Human-readable message for the event (no step prefix).\n:type description: str\n:param step: Optional executor task/step name (for example a Nomad task\n within the group).\n:type step: str | None",
"properties": {
"description": {
"title": "Description",
diff --git a/frontend/packages/api/specs/tasks.json b/frontend/packages/api/specs/tasks.json
index aca71ab41..484fd6e11 100644
--- a/frontend/packages/api/specs/tasks.json
+++ b/frontend/packages/api/specs/tasks.json
@@ -114,7 +114,7 @@
"type": "object"
},
"ExecutionEvent": {
- "description": "A single lifecycle event from a task executor (executor-agnostic shape).\n\n:param timestamp: When the event occurred (UTC).\n:type timestamp: UTCDatetime\n:param event_type: Executor-specific event category (e.g. Nomad task event type).\n:type event_type: str\n:param description: Human-readable message for the event (no step prefix).\n:type description: str\n:param step: Optional executor task/step name (e.g. Nomad task within the group).\n:type step: str | None",
+ "description": "A single lifecycle event from a task executor (executor-agnostic shape).\n\n:param timestamp: When the event occurred (UTC).\n:type timestamp: UTCDatetime\n:param event_type: Executor-specific event category (for example a Nomad\n task event type).\n:type event_type: str\n:param description: Human-readable message for the event (no step prefix).\n:type description: str\n:param step: Optional executor task/step name (for example a Nomad task\n within the group).\n:type step: str | None",
"properties": {
"description": {
"title": "Description",
@@ -963,7 +963,7 @@
"type": "object"
},
"TaskHistory": {
- "description": "Represent a task execution history.\n\n:param execution_request: The request that triggered the task execution.\n:type execution_request: TaskExecutionRequest\n:param status: The status of the task execution. Defaults to pending.\n:type status: TaskHistoryStatusEnum\n:param started_at: The datetime when the task execution started.\n:type started_at: UTCDatetime | None\n:param finished_at: The datetime when the task execution finished.\n:type finished_at: UTCDatetime | None\n:param anonymize_mask: The bitmask representing PII entities to be anonymized in\n logs and files generated by the execution. Defaults to None, meaning it uses\n the value defined in the associated task's :attr:`Task.anonymize_mask`.\n:type anonymize_mask: int | None\n:param task_id: The ID of the task associated with the execution.\n:type task_id: int\n:param task: The task associated with this execution history.\n:type task: Task\n:param sync_in_progress_started_at: Timestamp lock for a sync currently in progress.\n:type sync_in_progress_started_at: UTCDatetime | None\n:param log_allocation_epoch: Task-level high-water mark of the current Nomad\n allocation ``CreateIndex``, stamped whenever the log frontier is reset. The\n log writer consults it on the first-insert path (before any per-stream\n ``TaskHistoryLogState`` row exists) to discard writes from a superseded\n allocation. ``0`` is the legacy/unknown sentinel that is trusted\n unconditionally.\n:param executed_by: The user ID of the user who executed the task.\n:type executed_by: str | None",
+ "description": "Represent a task execution history.\n\n:param execution_request: The request that triggered the task execution.\n:type execution_request: TaskExecutionRequest\n:param status: The status of the task execution. Defaults to pending.\n:type status: TaskHistoryStatusEnum\n:param started_at: The datetime when the task execution started.\n:type started_at: UTCDatetime | None\n:param finished_at: The datetime when the task execution finished.\n:type finished_at: UTCDatetime | None\n:param anonymize_mask: The bitmask representing PII entities to be anonymized in\n logs and files generated by the execution. Defaults to None, meaning it uses\n the value defined in the associated task's :attr:`Task.anonymize_mask`.\n:type anonymize_mask: int | None\n:param task_id: The ID of the task associated with the execution.\n:type task_id: int\n:param task: The task associated with this execution history.\n:type task: Task\n:param sync_in_progress_started_at: Timestamp lock for a sync currently in progress.\n:type sync_in_progress_started_at: UTCDatetime | None\n:param log_producer_epoch: Task-level high-water mark of the current\n producer epoch (for example a Nomad allocation ``CreateIndex``), stamped\n whenever the log frontier is reset. The log writer consults it on the\n first-insert path (before any per-stream ``TaskHistoryLogState`` row\n exists) to discard writes from a superseded producer. ``0`` is the\n legacy/unknown sentinel that is trusted unconditionally.\n:param executed_by: The user ID of the user who executed the task.\n:type executed_by: str | None",
"properties": {
"anonymize_mask": {
"anyOf": [
@@ -1018,8 +1018,8 @@
],
"title": "Id"
},
- "log_allocation_epoch": {
- "title": "Log Allocation Epoch",
+ "log_producer_epoch": {
+ "title": "Log Producer Epoch",
"type": "integer"
},
"started_at": {
@@ -1071,7 +1071,7 @@
"id",
"execution_request",
"task_id",
- "log_allocation_epoch"
+ "log_producer_epoch"
],
"title": "TaskHistory",
"type": "object"
diff --git a/frontend/packages/api/src/generated/sep.ts b/frontend/packages/api/src/generated/sep.ts
index 312d417a6..8d7377bc0 100644
--- a/frontend/packages/api/src/generated/sep.ts
+++ b/frontend/packages/api/src/generated/sep.ts
@@ -3476,11 +3476,13 @@ export interface components {
*
* :param timestamp: When the event occurred (UTC).
* :type timestamp: UTCDatetime
- * :param event_type: Executor-specific event category (e.g. Nomad task event type).
+ * :param event_type: Executor-specific event category (for example a Nomad
+ * task event type).
* :type event_type: str
* :param description: Human-readable message for the event (no step prefix).
* :type description: str
- * :param step: Optional executor task/step name (e.g. Nomad task within the group).
+ * :param step: Optional executor task/step name (for example a Nomad task
+ * within the group).
* :type step: str | None
*/
ExecutionEvent: {
diff --git a/frontend/packages/api/src/generated/tasks.ts b/frontend/packages/api/src/generated/tasks.ts
index 9660ca48d..7f97a0f78 100644
--- a/frontend/packages/api/src/generated/tasks.ts
+++ b/frontend/packages/api/src/generated/tasks.ts
@@ -727,11 +727,13 @@ export interface components {
*
* :param timestamp: When the event occurred (UTC).
* :type timestamp: UTCDatetime
- * :param event_type: Executor-specific event category (e.g. Nomad task event type).
+ * :param event_type: Executor-specific event category (for example a Nomad
+ * task event type).
* :type event_type: str
* :param description: Human-readable message for the event (no step prefix).
* :type description: str
- * :param step: Optional executor task/step name (e.g. Nomad task within the group).
+ * :param step: Optional executor task/step name (for example a Nomad task
+ * within the group).
* :type step: str | None
*/
ExecutionEvent: {
@@ -1328,12 +1330,12 @@ export interface components {
* :type task: Task
* :param sync_in_progress_started_at: Timestamp lock for a sync currently in progress.
* :type sync_in_progress_started_at: UTCDatetime | None
- * :param log_allocation_epoch: Task-level high-water mark of the current Nomad
- * allocation ``CreateIndex``, stamped whenever the log frontier is reset. The
- * log writer consults it on the first-insert path (before any per-stream
- * ``TaskHistoryLogState`` row exists) to discard writes from a superseded
- * allocation. ``0`` is the legacy/unknown sentinel that is trusted
- * unconditionally.
+ * :param log_producer_epoch: Task-level high-water mark of the current
+ * producer epoch (for example a Nomad allocation ``CreateIndex``), stamped
+ * whenever the log frontier is reset. The log writer consults it on the
+ * first-insert path (before any per-stream ``TaskHistoryLogState`` row
+ * exists) to discard writes from a superseded producer. ``0`` is the
+ * legacy/unknown sentinel that is trusted unconditionally.
* :param executed_by: The user ID of the user who executed the task.
* :type executed_by: str | None
*/
@@ -1352,8 +1354,8 @@ export interface components {
finished_at?: string | null;
/** Id */
id: number | null;
- /** Log Allocation Epoch */
- log_allocation_epoch: number;
+ /** Log Producer Epoch */
+ log_producer_epoch: number;
/** Started At */
started_at?: string | null;
/** @default pending */
diff --git a/tests/app/sep/apps/framework/test_spec.py b/tests/app/sep/apps/framework/test_spec.py
index 70487ea80..dc9c9b076 100644
--- a/tests/app/sep/apps/framework/test_spec.py
+++ b/tests/app/sep/apps/framework/test_spec.py
@@ -57,8 +57,8 @@
CONNECTIVITY_META_PORT_KEY,
CONNECTIVITY_META_SERVICE_TYPE_KEY,
)
+from app.tasks.execution.executors.nomad.constants import RUN_SCRIPT_OUTPUT_FILES_PATH
from app.tasks.models import (
- RUN_SCRIPT_OUTPUT_FILES_PATH,
TaskBackendEnum,
TaskWrite,
)
diff --git a/tests/app/tasks/db/test_seed.py b/tests/app/tasks/db/test_seed.py
index 90d3d7141..135daf116 100644
--- a/tests/app/tasks/db/test_seed.py
+++ b/tests/app/tasks/db/test_seed.py
@@ -31,8 +31,10 @@
SYSTEM_PERIODIC_TASKS,
SYSTEM_TASKS,
)
-from app.tasks.models import (
+from app.tasks.execution.executors.nomad.constants import (
CHECK_NOMAD_CERT_EXPIRY_TASK_NAME,
+)
+from app.tasks.models import (
INTERNAL_TASK_NAMES,
SYNC_RUNNING_TASKS_TASK_NAME,
TaskBackendEnum,
diff --git a/tests/app/tasks/execution/executors/nomad/test_import_cycle.py b/tests/app/tasks/execution/executors/nomad/test_import_cycle.py
new file mode 100644
index 000000000..260dd7a04
--- /dev/null
+++ b/tests/app/tasks/execution/executors/nomad/test_import_cycle.py
@@ -0,0 +1,59 @@
+# Copyright (C) 2026 Percona LLC
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see .
+
+"""Regression tests for the Nomad executor package import cycle."""
+
+import subprocess
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[6]
+
+# Submodules that must load in a clean interpreter without importing
+# ``app.tasks.config`` first. ``constants`` is the light dependency-free probe;
+# ``models`` is the heavy end of the former cycle.
+_SUBMODULES = (
+ "app.tasks.execution.executors.nomad.constants",
+ "app.tasks.execution.executors.nomad.exceptions",
+ "app.tasks.execution.executors.nomad.models",
+)
+
+
+def test_nomad_package_submodules_import_without_config_first() -> None:
+ """Assert any Nomad executor submodule imports cleanly before tasks config.
+
+ Live entrypoints usually import ``app.tasks.config`` first, which hid a real
+ cycle: package ``__init__`` → ``models`` → ``config`` → package
+ ``NomadExecutor``. A fresh interpreter that touches a submodule first must
+ still succeed.
+ """
+ probe = (
+ "import importlib\n"
+ f"for name in {_SUBMODULES!r}:\n"
+ " importlib.import_module(name)\n"
+ "from app.tasks.execution.executors.nomad import NomadExecutor\n"
+ "assert NomadExecutor.__name__ == 'NomadExecutor'\n"
+ )
+ result = subprocess.run(
+ [sys.executable, "-c", probe],
+ cwd=REPO_ROOT,
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+ assert result.returncode == 0, (
+ f"Nomad package submodule import failed in a clean interpreter:\n"
+ f"stdout={result.stdout!r}\nstderr={result.stderr!r}"
+ )
diff --git a/tests/app/tasks/execution/executors/nomad/test_models.py b/tests/app/tasks/execution/executors/nomad/test_models.py
index 9450cebd4..6333f2f15 100644
--- a/tests/app/tasks/execution/executors/nomad/test_models.py
+++ b/tests/app/tasks/execution/executors/nomad/test_models.py
@@ -3249,8 +3249,8 @@ async def test_build_initial_log_offsets_skips_superseded_epoch_row(
new_bytes=b"old",
force_flush=True,
producer_offset_after=SEED_OFFSET,
- nomad_offset_after=SEED_OFFSET,
- allocation_epoch=SUPERSEDED_ALLOCATION_EPOCH,
+ producer_fetch_offset_after=SEED_OFFSET,
+ producer_epoch=SUPERSEDED_ALLOCATION_EPOCH,
)
offsets = await NomadExecutor._build_initial_log_offsets(
@@ -3273,8 +3273,8 @@ async def test_build_initial_log_offsets_seeds_matching_epoch_row(
new_bytes=b"cur",
force_flush=True,
producer_offset_after=SEED_OFFSET,
- nomad_offset_after=SEED_OFFSET,
- allocation_epoch=CURRENT_ALLOCATION_EPOCH,
+ producer_fetch_offset_after=SEED_OFFSET,
+ producer_epoch=CURRENT_ALLOCATION_EPOCH,
)
offsets = await NomadExecutor._build_initial_log_offsets(
@@ -3288,7 +3288,7 @@ async def test_build_initial_log_offsets_seeds_matching_epoch_row(
async def test_build_initial_log_offsets_seeds_legacy_epoch_zero_row(
self, session, created_task_with_history
):
- """Assert a legacy ``allocation_epoch == 0`` row is trusted for seeding."""
+ """Assert a legacy ``producer_epoch == 0`` row is trusted for seeding."""
history = created_task_with_history
await TaskHistoryLogWriter.append(
session,
diff --git a/tests/app/tasks/logs/test_log_eviction.py b/tests/app/tasks/logs/test_log_eviction.py
index 9b9724873..dcf62f57c 100644
--- a/tests/app/tasks/logs/test_log_eviction.py
+++ b/tests/app/tasks/logs/test_log_eviction.py
@@ -317,7 +317,7 @@ async def test_drain_preserves_persisted_offset_for_eviction(
persisted_before = state.persisted_offset
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history.id, new_allocation_epoch=REALLOCATION_EPOCH
+ session, history.id, new_producer_epoch=REALLOCATION_EPOCH
)
state = await TaskHistoryLogStateManager.get_for_stream(
diff --git a/tests/app/tasks/logs/test_log_writer.py b/tests/app/tasks/logs/test_log_writer.py
index ad38c15f2..e256c63d2 100644
--- a/tests/app/tasks/logs/test_log_writer.py
+++ b/tests/app/tasks/logs/test_log_writer.py
@@ -378,12 +378,12 @@ async def test_reset_producer_offsets_clears_db_and_allows_realloc_writes(
new_bytes=b"alloc-a content",
force_flush=True,
producer_offset_after=50_000,
- nomad_offset_after=50_000,
- allocation_epoch=ALLOCATION_EPOCH_OLD,
+ producer_fetch_offset_after=50_000,
+ producer_epoch=ALLOCATION_EPOCH_OLD,
)
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history.id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ session, history.id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
state = await TaskHistoryLogStateManager.get_for_stream(
@@ -391,8 +391,8 @@ async def test_reset_producer_offsets_clears_db_and_allows_realloc_writes(
)
assert state is not None
assert state.producer_offset == 0
- assert state.nomad_offset == 0
- assert state.allocation_epoch == ALLOCATION_EPOCH_NEW
+ assert state.producer_fetch_offset == 0
+ assert state.producer_epoch == ALLOCATION_EPOCH_NEW
persisted_after_alloc_a = state.persisted_offset
await TaskHistoryLogWriter.append(
@@ -403,23 +403,23 @@ async def test_reset_producer_offsets_clears_db_and_allows_realloc_writes(
new_bytes=b"alloc-b content",
force_flush=True,
producer_offset_after=ALLOC_B_PRODUCER_OFFSET,
- nomad_offset_after=len(b"alloc-b content"),
- allocation_epoch=ALLOCATION_EPOCH_NEW,
+ producer_fetch_offset_after=len(b"alloc-b content"),
+ producer_epoch=ALLOCATION_EPOCH_NEW,
)
state = await TaskHistoryLogStateManager.get_for_stream(
session, history.id, "run-script", TaskLogType.STDOUT
)
assert state is not None
assert state.producer_offset == ALLOC_B_PRODUCER_OFFSET
- assert state.nomad_offset == len(b"alloc-b content")
- assert state.allocation_epoch == ALLOCATION_EPOCH_NEW
+ assert state.producer_fetch_offset == len(b"alloc-b content")
+ assert state.producer_epoch == ALLOCATION_EPOCH_NEW
assert state.persisted_offset == persisted_after_alloc_a + len(b"alloc-b content")
chunks = await TaskHistoryLogManager.list_chunks_for_task(session, history.id)
assert [chunk.content for chunk in chunks] == ["alloc-a content", "alloc-b content"]
@pytest.mark.asyncio
-async def test_append_discards_write_from_older_allocation_epoch(
+async def test_append_discards_write_from_older_producer_epoch(
session: AsyncSession, created_task_with_history: TaskHistory
):
"""Assert an append whose epoch predates the committed row's epoch is dropped.
@@ -438,11 +438,11 @@ async def test_append_discards_write_from_older_allocation_epoch(
new_bytes=b"epoch-100",
force_flush=True,
producer_offset_after=len(b"epoch-100"),
- nomad_offset_after=len(b"epoch-100"),
- allocation_epoch=ALLOCATION_EPOCH_OLD,
+ producer_fetch_offset_after=len(b"epoch-100"),
+ producer_epoch=ALLOCATION_EPOCH_OLD,
)
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history.id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ session, history.id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
await TaskHistoryLogWriter.append(
@@ -453,17 +453,17 @@ async def test_append_discards_write_from_older_allocation_epoch(
new_bytes=b"stale-from-dead-alloc",
force_flush=True,
producer_offset_after=1_000,
- nomad_offset_after=1_000,
- allocation_epoch=ALLOCATION_EPOCH_OLD,
+ producer_fetch_offset_after=1_000,
+ producer_epoch=ALLOCATION_EPOCH_OLD,
)
state = await TaskHistoryLogStateManager.get_for_stream(
session, history.id, "run-script", TaskLogType.STDOUT
)
assert state is not None
- assert state.allocation_epoch == ALLOCATION_EPOCH_NEW
+ assert state.producer_epoch == ALLOCATION_EPOCH_NEW
assert state.producer_offset == 0
- assert state.nomad_offset == 0
+ assert state.producer_fetch_offset == 0
chunks = await TaskHistoryLogManager.list_chunks_for_task(session, history.id)
assert [chunk.content for chunk in chunks] == ["epoch-100"]
@@ -475,13 +475,13 @@ async def test_append_discards_write_from_older_allocation_epoch(
new_bytes=b"fresh-alloc",
force_flush=True,
producer_offset_after=len(b"fresh-alloc"),
- nomad_offset_after=len(b"fresh-alloc"),
- allocation_epoch=ALLOCATION_EPOCH_NEW,
+ producer_fetch_offset_after=len(b"fresh-alloc"),
+ producer_epoch=ALLOCATION_EPOCH_NEW,
)
state = await TaskHistoryLogStateManager.get_for_stream(
session, history.id, "run-script", TaskLogType.STDOUT
)
- assert state.allocation_epoch == ALLOCATION_EPOCH_NEW
+ assert state.producer_epoch == ALLOCATION_EPOCH_NEW
assert state.producer_offset == len(b"fresh-alloc")
chunks = await TaskHistoryLogManager.list_chunks_for_task(session, history.id)
assert [chunk.content for chunk in chunks] == ["epoch-100", "fresh-alloc"]
@@ -512,8 +512,8 @@ async def test_append_discard_guard_survives_version_retry(
new_bytes=b"seed",
force_flush=True,
producer_offset_after=len(b"seed"),
- nomad_offset_after=len(b"seed"),
- allocation_epoch=ALLOCATION_EPOCH_OLD,
+ producer_fetch_offset_after=len(b"seed"),
+ producer_epoch=ALLOCATION_EPOCH_OLD,
)
real_persist_state = TaskHistoryLogWriter._persist_state
@@ -523,7 +523,7 @@ async def racing_persist_state(**kwargs):
persist_calls["count"] += 1
if persist_calls["count"] == 1:
await TaskHistoryLogStateManager.reset_allocation_frontier(
- session, history.id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ session, history.id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
await session.commit()
return False
@@ -540,15 +540,15 @@ async def racing_persist_state(**kwargs):
stream=TaskLogType.STDOUT,
new_bytes=b"stale-through-retry",
producer_offset_after=1_000,
- nomad_offset_after=1_000,
- allocation_epoch=ALLOCATION_EPOCH_OLD,
+ producer_fetch_offset_after=1_000,
+ producer_epoch=ALLOCATION_EPOCH_OLD,
)
assert persist_calls["count"] == 1
state = await TaskHistoryLogStateManager.get_for_stream(
session, history.id, "run-script", TaskLogType.STDOUT
)
- assert state.allocation_epoch == ALLOCATION_EPOCH_NEW
+ assert state.producer_epoch == ALLOCATION_EPOCH_NEW
chunks = await TaskHistoryLogManager.list_chunks_for_task(session, history.id)
assert [chunk.content for chunk in chunks] == ["seed"]
@@ -569,7 +569,7 @@ async def test_append_discards_stale_first_insert_during_switch(
# The discard rolls back to release its lock, which expires the fixture row.
history_id = created_task_with_history.id
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history_id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ session, history_id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
await TaskHistoryLogWriter.append(
@@ -580,8 +580,8 @@ async def test_append_discards_stale_first_insert_during_switch(
new_bytes=b"stale-first-insert",
force_flush=True,
producer_offset_after=len(b"stale-first-insert"),
- nomad_offset_after=len(b"stale-first-insert"),
- allocation_epoch=ALLOCATION_EPOCH_OLD,
+ producer_fetch_offset_after=len(b"stale-first-insert"),
+ producer_epoch=ALLOCATION_EPOCH_OLD,
)
# Discard must roll back to free the row lock. Assert before any read below,
# which would autobegin a fresh transaction.
@@ -602,13 +602,13 @@ async def test_append_discards_stale_first_insert_during_switch(
new_bytes=b"current-alloc",
force_flush=True,
producer_offset_after=len(b"current-alloc"),
- nomad_offset_after=len(b"current-alloc"),
- allocation_epoch=ALLOCATION_EPOCH_NEW,
+ producer_fetch_offset_after=len(b"current-alloc"),
+ producer_epoch=ALLOCATION_EPOCH_NEW,
)
state = await TaskHistoryLogStateManager.get_for_stream(
session, history_id, "run-script", TaskLogType.STDOUT
)
- assert state.allocation_epoch == ALLOCATION_EPOCH_NEW
+ assert state.producer_epoch == ALLOCATION_EPOCH_NEW
assert state.producer_offset == len(b"current-alloc")
chunks = await TaskHistoryLogManager.list_chunks_for_task(session, history_id)
assert [chunk.content for chunk in chunks] == ["current-alloc"]
@@ -626,7 +626,7 @@ async def test_append_first_insert_accepts_epoch_at_or_above_hwm(
"""
history = created_task_with_history
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history.id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ session, history.id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
await TaskHistoryLogWriter.append(
@@ -637,14 +637,14 @@ async def test_append_first_insert_accepts_epoch_at_or_above_hwm(
new_bytes=b"live-first-insert",
force_flush=True,
producer_offset_after=len(b"live-first-insert"),
- nomad_offset_after=len(b"live-first-insert"),
- allocation_epoch=ALLOCATION_EPOCH_NEW,
+ producer_fetch_offset_after=len(b"live-first-insert"),
+ producer_epoch=ALLOCATION_EPOCH_NEW,
)
state = await TaskHistoryLogStateManager.get_for_stream(
session, history.id, "run-script", TaskLogType.STDOUT
)
assert state is not None
- assert state.allocation_epoch == ALLOCATION_EPOCH_NEW
+ assert state.producer_epoch == ALLOCATION_EPOCH_NEW
chunks = await TaskHistoryLogManager.list_chunks_for_task(session, history.id)
assert [chunk.content for chunk in chunks] == ["live-first-insert"]
@@ -668,14 +668,14 @@ async def test_append_first_insert_without_hwm_accepts_write(
new_bytes=b"first-alloc",
force_flush=True,
producer_offset_after=len(b"first-alloc"),
- nomad_offset_after=len(b"first-alloc"),
- allocation_epoch=ALLOCATION_EPOCH_LIVE,
+ producer_fetch_offset_after=len(b"first-alloc"),
+ producer_epoch=ALLOCATION_EPOCH_LIVE,
)
state = await TaskHistoryLogStateManager.get_for_stream(
session, history.id, "run-script", TaskLogType.STDOUT
)
assert state is not None
- assert state.allocation_epoch == ALLOCATION_EPOCH_LIVE
+ assert state.producer_epoch == ALLOCATION_EPOCH_LIVE
chunks = await TaskHistoryLogManager.list_chunks_for_task(session, history.id)
assert [chunk.content for chunk in chunks] == ["first-alloc"]
@@ -706,7 +706,7 @@ async def racing_persist_state(**kwargs):
persist_calls["count"] += 1
if persist_calls["count"] == 1:
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history_id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ session, history_id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
return False
return await real_persist_state(**kwargs)
@@ -722,8 +722,8 @@ async def racing_persist_state(**kwargs):
stream=TaskLogType.STDOUT,
new_bytes=b"stale-through-first-insert",
producer_offset_after=len(b"stale-through-first-insert"),
- nomad_offset_after=len(b"stale-through-first-insert"),
- allocation_epoch=ALLOCATION_EPOCH_OLD,
+ producer_fetch_offset_after=len(b"stale-through-first-insert"),
+ producer_epoch=ALLOCATION_EPOCH_OLD,
)
assert persist_calls["count"] == 1
@@ -741,7 +741,7 @@ async def test_drain_does_not_regress_high_water_mark_on_out_of_order_reset(
):
"""Assert an out-of-order drain with a smaller epoch never lowers the mark.
- Regression for the monotonicity guard in ``bump_log_allocation_epoch``: the
+ Regression for the monotonicity guard in ``bump_log_producer_epoch``: the
task-level high-water mark must only advance. A stale drain carrying a lower
``CreateIndex`` than the current mark is a no-op, so a superseded-allocation
first-insert stays discarded instead of being re-accepted after the mark is
@@ -749,19 +749,19 @@ async def test_drain_does_not_regress_high_water_mark_on_out_of_order_reset(
"""
history_id = created_task_with_history.id
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history_id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ session, history_id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
assert (
- await TaskHistoryManager.get_log_allocation_epoch(session, history_id)
+ await TaskHistoryManager.get_log_producer_epoch(session, history_id)
== ALLOCATION_EPOCH_NEW
)
# A late drain from the superseded allocation carries the smaller epoch.
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history_id, new_allocation_epoch=ALLOCATION_EPOCH_OLD
+ session, history_id, new_producer_epoch=ALLOCATION_EPOCH_OLD
)
assert (
- await TaskHistoryManager.get_log_allocation_epoch(session, history_id)
+ await TaskHistoryManager.get_log_producer_epoch(session, history_id)
== ALLOCATION_EPOCH_NEW
)
@@ -777,8 +777,8 @@ async def test_drain_does_not_regress_high_water_mark_on_out_of_order_reset(
new_bytes=b"mid-epoch-stale",
force_flush=True,
producer_offset_after=len(b"mid-epoch-stale"),
- nomad_offset_after=len(b"mid-epoch-stale"),
- allocation_epoch=mid_epoch,
+ producer_fetch_offset_after=len(b"mid-epoch-stale"),
+ producer_epoch=mid_epoch,
)
state = await TaskHistoryLogStateManager.get_for_stream(
session, history_id, "run-script", TaskLogType.STDOUT
@@ -802,7 +802,7 @@ async def test_append_discards_stale_first_insert_across_both_streams(
"""
history_id = created_task_with_history.id
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history_id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ session, history_id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
for stream in (TaskLogType.STDOUT, TaskLogType.STDERR):
@@ -814,8 +814,8 @@ async def test_append_discards_stale_first_insert_across_both_streams(
new_bytes=b"stale-" + stream.value.encode("utf-8"),
force_flush=True,
producer_offset_after=len(b"stale-" + stream.value.encode("utf-8")),
- nomad_offset_after=len(b"stale-" + stream.value.encode("utf-8")),
- allocation_epoch=ALLOCATION_EPOCH_OLD,
+ producer_fetch_offset_after=len(b"stale-" + stream.value.encode("utf-8")),
+ producer_epoch=ALLOCATION_EPOCH_OLD,
)
assert (
await TaskHistoryLogStateManager.get_for_stream(
@@ -835,14 +835,14 @@ async def test_append_discards_stale_first_insert_across_both_streams(
new_bytes=payload,
force_flush=True,
producer_offset_after=len(payload),
- nomad_offset_after=len(payload),
- allocation_epoch=ALLOCATION_EPOCH_NEW,
+ producer_fetch_offset_after=len(payload),
+ producer_epoch=ALLOCATION_EPOCH_NEW,
)
state = await TaskHistoryLogStateManager.get_for_stream(
session, history_id, "run-script", stream
)
assert state is not None
- assert state.allocation_epoch == ALLOCATION_EPOCH_NEW
+ assert state.producer_epoch == ALLOCATION_EPOCH_NEW
assert state.producer_offset == len(payload)
@@ -850,7 +850,7 @@ async def test_append_discards_stale_first_insert_across_both_streams(
async def test_append_legacy_epoch_zero_row_accepts_live_write(
session: AsyncSession, created_task_with_history: TaskHistory
):
- """Assert a legacy ``allocation_epoch == 0`` row is stamped by the next write.
+ """Assert a legacy ``producer_epoch == 0`` row is stamped by the next write.
Pre-migration rows carry the ``0`` sentinel; the discard guard must treat
them as trusted and let the first live write advance the epoch to the
@@ -869,7 +869,7 @@ async def test_append_legacy_epoch_zero_row_accepts_live_write(
state = await TaskHistoryLogStateManager.get_for_stream(
session, history.id, "run-script", TaskLogType.STDOUT
)
- assert state.allocation_epoch == 0
+ assert state.producer_epoch == 0
await TaskHistoryLogWriter.append(
session,
@@ -879,13 +879,13 @@ async def test_append_legacy_epoch_zero_row_accepts_live_write(
new_bytes=b"live-bytes",
force_flush=True,
producer_offset_after=len(b"legacy-bytes") + len(b"live-bytes"),
- nomad_offset_after=len(b"legacy-bytes") + len(b"live-bytes"),
- allocation_epoch=ALLOCATION_EPOCH_LIVE,
+ producer_fetch_offset_after=len(b"legacy-bytes") + len(b"live-bytes"),
+ producer_epoch=ALLOCATION_EPOCH_LIVE,
)
state = await TaskHistoryLogStateManager.get_for_stream(
session, history.id, "run-script", TaskLogType.STDOUT
)
- assert state.allocation_epoch == ALLOCATION_EPOCH_LIVE
+ assert state.producer_epoch == ALLOCATION_EPOCH_LIVE
chunks = await TaskHistoryLogManager.list_chunks_for_task(session, history.id)
assert [chunk.content for chunk in chunks] == ["legacy-bytes", "live-bytes"]
@@ -908,8 +908,8 @@ async def test_append_non_nomad_caller_leaves_frontier_columns_zero(
session, history.id, "execution", TaskLogType.STDOUT
)
assert state is not None
- assert state.nomad_offset == 0
- assert state.allocation_epoch == 0
+ assert state.producer_fetch_offset == 0
+ assert state.producer_epoch == 0
chunks = await TaskHistoryLogManager.list_chunks_for_task(session, history.id)
assert [chunk.content for chunk in chunks] == ["celery-output"]
@@ -1053,7 +1053,7 @@ async def test_drain_and_reset_flushes_staging_before_zeroing_producer_offset(
assert chunks == []
await TaskHistoryLogWriter.drain_and_reset_allocation_frontier(
- session, history.id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ session, history.id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
state = await TaskHistoryLogStateManager.get_for_stream(
@@ -1094,8 +1094,8 @@ async def test_first_insert_lock_serialises_reset_on_postgres(
``with_for_update()`` is a no-op on SQLite, so the rest of this module proves
the epoch-discard *behaviour* but never the row-lock *ordering* it rests on.
Here two independent PostgreSQL-bound sessions race: the holder takes the
- first-insert lock via ``get_log_allocation_epoch(for_update=True)`` and keeps
- its transaction open; the resetter's ``bump_log_allocation_epoch`` + commit
+ first-insert lock via ``get_log_producer_epoch(for_update=True)`` and keeps
+ its transaction open; the resetter's ``bump_log_producer_epoch`` + commit
(the frontier reset) must block until the holder ends, then land — proving the
two serialise on the ``TaskHistory`` row rather than racing.
"""
@@ -1113,14 +1113,14 @@ async def test_first_insert_lock_serialises_reset_on_postgres(
async with maker() as holder, maker() as resetter:
# Holder takes the first-insert lock and keeps its transaction open.
- locked_epoch = await TaskHistoryManager.get_log_allocation_epoch(
+ locked_epoch = await TaskHistoryManager.get_log_producer_epoch(
holder, history_id, for_update=True
)
assert locked_epoch == 0
async def _reset() -> None:
- await TaskHistoryManager.bump_log_allocation_epoch(
- resetter, history_id, new_allocation_epoch=ALLOCATION_EPOCH_NEW
+ await TaskHistoryManager.bump_log_producer_epoch(
+ resetter, history_id, new_producer_epoch=ALLOCATION_EPOCH_NEW
)
await resetter.commit()
@@ -1136,9 +1136,7 @@ async def _reset() -> None:
await asyncio.wait_for(reset_task, timeout=RESET_RELEASE_TIMEOUT_SEC)
async with maker() as verify:
- epoch = await TaskHistoryManager.get_log_allocation_epoch(
- verify, history_id
- )
+ epoch = await TaskHistoryManager.get_log_producer_epoch(verify, history_id)
assert epoch == ALLOCATION_EPOCH_NEW
finally:
async with postgres_engine.begin() as conn:
diff --git a/tests/app/tasks/migrations/test_rename_log_cursor_columns.py b/tests/app/tasks/migrations/test_rename_log_cursor_columns.py
new file mode 100644
index 000000000..25295ee82
--- /dev/null
+++ b/tests/app/tasks/migrations/test_rename_log_cursor_columns.py
@@ -0,0 +1,137 @@
+# Copyright (C) 2026 Percona LLC
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see .
+
+"""Tests for the executor-neutral log-cursor column rename migration."""
+
+from pathlib import Path
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine, inspect
+
+from app.tasks.config import tasks_settings
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+ALEMBIC_INI = REPO_ROOT / "alembic.ini"
+
+_PRE_RENAME_REVISION = "27a11549ef43"
+_RENAME_REVISION = "c8e4a2b91f70"
+
+_FETCH_OFFSET = 4096
+_PRODUCER_EPOCH = 42
+_LOG_PRODUCER_EPOCH = 99
+
+
+@pytest.fixture
+def tasks_alembic_config(tmp_path, monkeypatch):
+ """Yield an Alembic ``Config`` and sync URL pointing at a temp SQLite file."""
+ db_path = tmp_path / "test_tasks_rename.sqlite"
+ sync_url = f"sqlite:///{db_path}"
+
+ monkeypatch.setattr(tasks_settings.DATABASE, "HOST", "")
+ monkeypatch.setattr(tasks_settings.DATABASE, "NAME", str(db_path))
+
+ cfg = Config(str(ALEMBIC_INI), ini_section="tasks")
+ return cfg, sync_url
+
+
+def test_rename_preserves_values_and_downgrades(tasks_alembic_config):
+ """Assert the rename migration keeps values and restores old names on downgrade."""
+ cfg, sync_url = tasks_alembic_config
+ command.upgrade(cfg, _PRE_RENAME_REVISION)
+
+ engine = create_engine(sync_url)
+ try:
+ with engine.begin() as conn:
+ conn.exec_driver_sql(
+ "INSERT INTO task "
+ "(created_at, name, data, backend, owner, is_template, protected, "
+ "alert_on_fail, anonymize_mask) "
+ "VALUES ('2026-01-01 00:00:00', 't', '{}', 'nomad', 'ANY', 0, 0, "
+ "0, 0)"
+ )
+ conn.exec_driver_sql(
+ "INSERT INTO taskhistory "
+ "(created_at, task_id, execution_request, status, "
+ "log_allocation_epoch) "
+ "VALUES ('2026-01-01 00:00:00', 1, '{}', 'pending', ?)",
+ (_LOG_PRODUCER_EPOCH,),
+ )
+ conn.exec_driver_sql(
+ "INSERT INTO taskhistory_log_state "
+ "(created_at, task_history_id, source, stream, persisted_offset, "
+ "producer_offset, nomad_offset, allocation_epoch, staging, "
+ "staging_updated_at, version) "
+ "VALUES ('2026-01-01 00:00:00', 1, 'run-script', 'STDOUT', 0, 0, "
+ "?, ?, X'', '2026-01-01 00:00:00', 0)",
+ (_FETCH_OFFSET, _PRODUCER_EPOCH),
+ )
+ finally:
+ engine.dispose()
+
+ command.upgrade(cfg, _RENAME_REVISION)
+
+ engine = create_engine(sync_url)
+ try:
+ with engine.begin() as conn:
+ state_cols = {
+ c["name"] for c in inspect(conn).get_columns("taskhistory_log_state")
+ }
+ history_cols = {c["name"] for c in inspect(conn).get_columns("taskhistory")}
+ assert "producer_fetch_offset" in state_cols
+ assert "producer_epoch" in state_cols
+ assert "nomad_offset" not in state_cols
+ assert "allocation_epoch" not in state_cols
+ assert "log_producer_epoch" in history_cols
+ assert "log_allocation_epoch" not in history_cols
+
+ state = conn.exec_driver_sql(
+ "SELECT producer_fetch_offset, producer_epoch "
+ "FROM taskhistory_log_state"
+ ).one()
+ history = conn.exec_driver_sql(
+ "SELECT log_producer_epoch FROM taskhistory"
+ ).one()
+ assert state.producer_fetch_offset == _FETCH_OFFSET
+ assert state.producer_epoch == _PRODUCER_EPOCH
+ assert history.log_producer_epoch == _LOG_PRODUCER_EPOCH
+ finally:
+ engine.dispose()
+
+ command.downgrade(cfg, _PRE_RENAME_REVISION)
+
+ engine = create_engine(sync_url)
+ try:
+ with engine.begin() as conn:
+ state_cols = {
+ c["name"] for c in inspect(conn).get_columns("taskhistory_log_state")
+ }
+ history_cols = {c["name"] for c in inspect(conn).get_columns("taskhistory")}
+ assert "nomad_offset" in state_cols
+ assert "allocation_epoch" in state_cols
+ assert "log_allocation_epoch" in history_cols
+
+ state = conn.exec_driver_sql(
+ "SELECT nomad_offset, allocation_epoch FROM taskhistory_log_state"
+ ).one()
+ history = conn.exec_driver_sql(
+ "SELECT log_allocation_epoch FROM taskhistory"
+ ).one()
+ assert state.nomad_offset == _FETCH_OFFSET
+ assert state.allocation_epoch == _PRODUCER_EPOCH
+ assert history.log_allocation_epoch == _LOG_PRODUCER_EPOCH
+ finally:
+ engine.dispose()
diff --git a/tests/app/tasks/migrations/test_taskhistory_log_nomad_cursor.py b/tests/app/tasks/migrations/test_taskhistory_log_nomad_cursor.py
index e12938bea..5b50ab742 100644
--- a/tests/app/tasks/migrations/test_taskhistory_log_nomad_cursor.py
+++ b/tests/app/tasks/migrations/test_taskhistory_log_nomad_cursor.py
@@ -27,10 +27,11 @@
REPO_ROOT = Path(__file__).resolve().parents[4]
ALEMBIC_INI = REPO_ROOT / "alembic.ini"
-# The merged head immediately before nomad_offset / allocation_epoch are added.
+# The merged head immediately before nomad_offset / allocation_epoch are added
+# (later renamed to producer_fetch_offset / producer_epoch).
_PRE_NOMAD_CURSOR_REVISION = "f028a195fbda"
# An anonymized-stream row: producer_offset diverges from the true raw offset,
-# so the backfill (nomad_offset = producer_offset) is the documented
+# so the backfill (originally nomad_offset = producer_offset) is the documented
# approximation rather than an exact seed.
_ANONYMIZED_PRODUCER_OFFSET = 4096
@@ -56,8 +57,10 @@ def tasks_alembic_config(tmp_path, monkeypatch):
return cfg, sync_url
-def test_backfill_seeds_nomad_offset_from_producer_offset(tasks_alembic_config):
- """Assert the upgrade backfills nomad_offset from producer_offset for in-flight rows."""
+def test_backfill_seeds_producer_fetch_offset_from_producer_offset(
+ tasks_alembic_config,
+):
+ """Assert the upgrade backfills producer_fetch_offset from producer_offset for in-flight rows."""
cfg, sync_url = tasks_alembic_config
command.upgrade(cfg, _PRE_NOMAD_CURSOR_REVISION)
@@ -74,10 +77,10 @@ def test_backfill_seeds_nomad_offset_from_producer_offset(tasks_alembic_config):
try:
with engine.begin() as conn:
row = conn.exec_driver_sql(
- "SELECT nomad_offset, allocation_epoch FROM taskhistory_log_state"
+ "SELECT producer_fetch_offset, producer_epoch FROM taskhistory_log_state"
).one()
- assert row.nomad_offset == _ANONYMIZED_PRODUCER_OFFSET
- assert row.allocation_epoch == 0
+ assert row.producer_fetch_offset == _ANONYMIZED_PRODUCER_OFFSET
+ assert row.producer_epoch == 0
finally:
engine.dispose()
@@ -92,9 +95,9 @@ def test_new_columns_default_zero_on_fresh_insert(tasks_alembic_config):
with engine.begin() as conn:
conn.exec_driver_sql(_INSERT_STATE_ROW, (0,))
row = conn.exec_driver_sql(
- "SELECT nomad_offset, allocation_epoch FROM taskhistory_log_state"
+ "SELECT producer_fetch_offset, producer_epoch FROM taskhistory_log_state"
).one()
- assert row.nomad_offset == 0
- assert row.allocation_epoch == 0
+ assert row.producer_fetch_offset == 0
+ assert row.producer_epoch == 0
finally:
engine.dispose()
diff --git a/tests/app/tasks/test_routes.py b/tests/app/tasks/test_routes.py
index f77a311b1..28db9b7dc 100644
--- a/tests/app/tasks/test_routes.py
+++ b/tests/app/tasks/test_routes.py
@@ -42,6 +42,7 @@
from app.tasks.connectivity.service import _cached_check_connectivity
from app.tasks.crud import TaskHistoryLogManager, TaskHistoryManager, TaskManager
from app.tasks.deps import get_request_executor, get_session
+from app.tasks.execution.executors.nomad.constants import RUN_SCRIPT_OUTPUT_FILES_PATH
from app.tasks.execution.executors.nomad.exceptions import AllocationNotFoundError
from app.tasks.execution.models import BaseExecutor
from app.tasks.logs.log_writer import TaskHistoryLogWriter
@@ -49,7 +50,6 @@
from app.tasks.models import (
DispatchLock,
ExecutionEvent,
- RUN_SCRIPT_OUTPUT_FILES_PATH,
SYSTEM_USER,
Task,
TaskBackendEnum,