You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by sb...@apache.org on 2026/07/28 17:40:00 UTC
(tooling-trusted-releases) branch sbp updated (15edfa9b -> 70f821b1)
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a change to branch sbp
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
discard 15edfa9b Record failed and broken checks as exception results
discard 5c5bc8d5 Add a broken status to indicate tasks which may be retried
add f50486b8 Fix on dist watcher for whole-directory adds
add ace7218a #1461 - rearrange catalog
new 70f821b1 Add a broken status to indicate tasks which may be retried
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (15edfa9b)
\
N -- N -- N refs/heads/sbp (70f821b1)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
atr/catalog_site.py | 144 +++++++++++++++++----
atr/get/checks.py | 12 +-
atr/get/compose.py | 16 +++
atr/manager.py | 39 +++---
atr/render.py | 28 ++++
atr/svn/__init__.py | 7 +
atr/svn/catalog.py | 60 +++++++--
atr/tasks/task.py | 116 +++--------------
atr/templates/catalog_site/_project_back_link.html | 23 ++--
atr/templates/catalog_site/catalog_site_base.html | 1 +
atr/templates/catalog_site/committee.html | 47 -------
atr/templates/catalog_site/index.html | 2 +-
atr/templates/catalog_site/project.html | 60 ++++++---
atr/templates/check-selected-path-table.html | 1 +
atr/worker.py | 84 +++++-------
tests/unit/test_checks_table.py | 43 ++++--
tests/unit/test_manager_terminate.py | 78 -----------
tests/unit/test_task_finalise.py | 129 ------------------
tests/unit/test_worker_complete.py | 107 ---------------
tests/unit/test_worker_defer.py | 10 +-
20 files changed, 395 insertions(+), 612 deletions(-)
delete mode 100644 atr/templates/catalog_site/committee.html
delete mode 100644 tests/unit/test_manager_terminate.py
delete mode 100644 tests/unit/test_task_finalise.py
delete mode 100644 tests/unit/test_worker_complete.py
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org
(tooling-trusted-releases) 01/01: Add a broken status to indicate tasks which may be retried
Posted by sb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a commit to branch sbp
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
commit 70f821b1f0f1bae73ecc6dfecf9f29db940f8cb9
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Tue Jul 28 17:52:32 2026 +0100
Add a broken status to indicate tasks which may be retried
---
atr/admin/__init__.py | 1 +
atr/db/interaction.py | 2 +-
atr/manager.py | 4 +-
atr/models/sql.py | 7 +-
atr/tasks/task.py | 10 +-
atr/worker.py | 15 ++-
migrations/versions/0113_2026.07.28_dd069840.py | 124 ++++++++++++++++++++++++
tests/unit/test_worker_defer.py | 27 ++++--
8 files changed, 174 insertions(+), 16 deletions(-)
diff --git a/atr/admin/__init__.py b/atr/admin/__init__.py
index 5c456a67..29328388 100644
--- a/atr/admin/__init__.py
+++ b/atr/admin/__init__.py
@@ -1927,6 +1927,7 @@ async def tasks_recent(_session: web.Committer, _tasks_recent: Literal["tasks/re
sql.TaskStatus.ACTIVE: ".table-info",
sql.TaskStatus.COMPLETED: ".table-success",
sql.TaskStatus.FAILED: ".table-danger",
+ sql.TaskStatus.BROKEN: ".table-warning",
}.get(task.status, "")
if (task.started is not None) and (task.completed is not None):
took_seconds = (task.completed - task.started).total_seconds()
diff --git a/atr/db/interaction.py b/atr/db/interaction.py
index 46911de4..59b607d1 100644
--- a/atr/db/interaction.py
+++ b/atr/db/interaction.py
@@ -1076,7 +1076,7 @@ async def wait_for_task(
return False
for _attempt in range(timeout_s * 10):
await data.refresh(t)
- if t.status == sql.TaskStatus.FAILED:
+ if t.status in (sql.TaskStatus.FAILED, sql.TaskStatus.BROKEN):
raise InteractionError(f"Task {task.id} failed with error {t.error}")
if t.status == desired_status:
return True
diff --git a/atr/manager.py b/atr/manager.py
index ab450bbe..1b2e64cb 100644
--- a/atr/manager.py
+++ b/atr/manager.py
@@ -325,7 +325,9 @@ class WorkerManager:
.select_from(sql.Task)
.where(
sql.Task.pid == pid,
- via(sql.Task.status).in_([sql.TaskStatus.COMPLETED, sql.TaskStatus.FAILED]),
+ via(sql.Task.status).in_(
+ [sql.TaskStatus.COMPLETED, sql.TaskStatus.FAILED, sql.TaskStatus.BROKEN]
+ ),
via(sql.Task.started) >= worker.started,
)
)
diff --git a/atr/models/sql.py b/atr/models/sql.py
index 49a318fd..1adb4ff0 100644
--- a/atr/models/sql.py
+++ b/atr/models/sql.py
@@ -227,6 +227,7 @@ class TaskStatus(enum.StrEnum):
ACTIVE = "active"
COMPLETED = "completed"
FAILED = "failed"
+ BROKEN = "broken"
class TaskType(enum.StrEnum):
@@ -783,8 +784,8 @@ class Task(sqlmodel.SQLModel, table=True):
sqlalchemy.Index("ix_task_status_added", "status", "added"),
# Ensure valid status transitions:
# - QUEUED can transition to ACTIVE
- # - ACTIVE can transition to COMPLETED or FAILED
- # - COMPLETED and FAILED are terminal states
+ # - ACTIVE can transition to COMPLETED, FAILED, or BROKEN
+ # - COMPLETED, FAILED, and BROKEN are terminal states
sqlalchemy.CheckConstraint(
"""
(
@@ -796,6 +797,8 @@ class Task(sqlmodel.SQLModel, table=True):
OR (status = 'COMPLETED' AND completed IS NOT NULL AND result IS NOT NULL)
-- ACTIVE -> FAILED requires setting completed time and error (result optional)
OR (status = 'FAILED' AND completed IS NOT NULL AND error IS NOT NULL)
+ -- ACTIVE -> BROKEN requires setting completed time and error (result optional)
+ OR (status = 'BROKEN' AND completed IS NOT NULL AND error IS NOT NULL)
)
""",
name="valid_task_status_transitions",
diff --git a/atr/tasks/task.py b/atr/tasks/task.py
index db64a768..62bc26ae 100644
--- a/atr/tasks/task.py
+++ b/atr/tasks/task.py
@@ -17,7 +17,7 @@
from __future__ import annotations
-from typing import Final
+from typing import Any, Final
import atr.constants as constants
import atr.log as log
@@ -29,6 +29,7 @@ QUEUED: Final = sql.TaskStatus.QUEUED
ACTIVE: Final = sql.TaskStatus.ACTIVE
COMPLETED: Final = sql.TaskStatus.COMPLETED
FAILED: Final = sql.TaskStatus.FAILED
+BROKEN: Final = sql.TaskStatus.BROKEN
# The recurring tasks reschedule themselves and carry no lasting per-run value, so on
# success they are logged to a file and dropped rather than kept in the table.
@@ -50,6 +51,13 @@ TASK_TYPE_TIMEOUT_SECONDS: Final[dict[sql.TaskType, int]] = {
}
+class CheckRetryableError(Exception):
+ def __init__(self, message: str, data: Any = None) -> None:
+ super().__init__(message)
+ self.message = message
+ self.data = data
+
+
class DeferredError(Exception):
"""Raised by a handler to send its own task back to the queue for a later attempt."""
diff --git a/atr/worker.py b/atr/worker.py
index 97190bc9..7896be1b 100644
--- a/atr/worker.py
+++ b/atr/worker.py
@@ -190,10 +190,6 @@ def _setup_logging() -> None:
)
-def _task_completed_log(record: dict[str, Any]) -> None:
- logging.getLogger(_TASK_LOG_LOGGER).info(json.dumps(record, allow_nan=False))
-
-
def _task_args_for_log(
task_type: str | sql.TaskType, task_args: list[str] | dict[str, Any]
) -> list[str] | dict[str, Any]:
@@ -204,6 +200,10 @@ def _task_args_for_log(
}
+def _task_completed_log(record: dict[str, Any]) -> None:
+ logging.getLogger(_TASK_LOG_LOGGER).info(json.dumps(record, allow_nan=False))
+
+
def _task_completed_record(
task_obj: sql.Task, result: results.Results | None, completed: datetime.datetime
) -> dict[str, Any]:
@@ -346,6 +346,11 @@ async def _task_process(task_id: int, task_type: str, task_args: list[str] | dic
log.info(f"Task {task_id} ({task_type}) deferred, re-queued for a later attempt")
await _task_defer(task_id)
return
+ except task.CheckRetryableError as e:
+ task_results = None
+ status = task.BROKEN
+ log.error(f"Task {task_id} failed with a retryable error: {e}")
+ error = str(e)
except Exception as e:
task_results = None
status = task.FAILED
@@ -377,7 +382,7 @@ async def _task_result_process(
task_obj.completed = completed
task_obj.result = task_results
- if status == task.FAILED:
+ if status in (task.FAILED, task.BROKEN):
normalised_error = ((error or "").strip()) or "unknown error"
task_obj.error = normalised_error
notify_args = (
diff --git a/migrations/versions/0113_2026.07.28_dd069840.py b/migrations/versions/0113_2026.07.28_dd069840.py
new file mode 100644
index 00000000..9844d94a
--- /dev/null
+++ b/migrations/versions/0113_2026.07.28_dd069840.py
@@ -0,0 +1,124 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Add the BROKEN task status for retryable failures
+
+Revision ID: 0113_2026.07.28_dd069840
+Revises: 0112_2026.07.22_d374db02
+Create Date: 2026-07-28 13:34:48.150920+00:00
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# Revision identifiers, used by Alembic
+revision: str = "0113_2026.07.28_dd069840"
+down_revision: str | None = "0112_2026.07.22_d374db02"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_CHECK_WITHOUT_BROKEN = """
+ (
+ -- Initial state is always valid
+ status = 'QUEUED'
+ -- QUEUED -> ACTIVE requires setting started time and pid
+ OR (status = 'ACTIVE' AND started IS NOT NULL AND pid IS NOT NULL)
+ -- ACTIVE -> COMPLETED requires setting completed time and result
+ OR (status = 'COMPLETED' AND completed IS NOT NULL AND result IS NOT NULL)
+ -- ACTIVE -> FAILED requires setting completed time and error (result optional)
+ OR (status = 'FAILED' AND completed IS NOT NULL AND error IS NOT NULL)
+ )
+ """
+
+_CHECK_WITH_BROKEN = """
+ (
+ -- Initial state is always valid
+ status = 'QUEUED'
+ -- QUEUED -> ACTIVE requires setting started time and pid
+ OR (status = 'ACTIVE' AND started IS NOT NULL AND pid IS NOT NULL)
+ -- ACTIVE -> COMPLETED requires setting completed time and result
+ OR (status = 'COMPLETED' AND completed IS NOT NULL AND result IS NOT NULL)
+ -- ACTIVE -> FAILED requires setting completed time and error (result optional)
+ OR (status = 'FAILED' AND completed IS NOT NULL AND error IS NOT NULL)
+ -- ACTIVE -> BROKEN requires setting completed time and error (result optional)
+ OR (status = 'BROKEN' AND completed IS NOT NULL AND error IS NOT NULL)
+ )
+ """
+
+
+def upgrade() -> None:
+ op.execute(
+ "UPDATE task SET completed = COALESCE(completed, started, added)"
+ " WHERE status IN ('COMPLETED', 'FAILED') AND completed IS NULL"
+ )
+ op.execute("UPDATE task SET result = 'null' WHERE status = 'COMPLETED' AND result IS NULL")
+ op.execute("UPDATE task SET error = 'unknown error' WHERE status = 'FAILED' AND error IS NULL")
+ op.execute(
+ "UPDATE task SET status = 'FAILED', completed = COALESCE(completed, started, added),"
+ " error = 'Task was active with no recorded worker'"
+ " WHERE status = 'ACTIVE' AND (started IS NULL OR pid IS NULL)"
+ )
+ with op.batch_alter_table("task", schema=None, copy_from=_task_table()) as batch_op:
+ batch_op.create_check_constraint(
+ batch_op.f("ck_task_valid_task_status_transitions"),
+ _CHECK_WITH_BROKEN,
+ )
+
+
+def downgrade() -> None:
+ op.execute("UPDATE task SET status = 'FAILED' WHERE status = 'BROKEN'")
+ with op.batch_alter_table("task", schema=None, copy_from=_task_table()) as batch_op:
+ batch_op.create_check_constraint(
+ batch_op.f("ck_task_valid_task_status_transitions"),
+ _CHECK_WITHOUT_BROKEN,
+ )
+
+
+def _task_table() -> sa.Table:
+ return sa.Table(
+ "task",
+ sa.MetaData(),
+ sa.Column("id", sa.Integer(), nullable=False),
+ sa.Column("status", sa.String(9), nullable=False),
+ sa.Column("task_type", sa.String(23), nullable=False),
+ sa.Column("task_args", sa.JSON(), nullable=True),
+ sa.Column("added", sa.TIMESTAMP(), nullable=False),
+ sa.Column("started", sa.TIMESTAMP(), nullable=True),
+ sa.Column("pid", sa.Integer(), nullable=True),
+ sa.Column("completed", sa.TIMESTAMP(), nullable=True),
+ sa.Column("result", sa.JSON(), nullable=True),
+ sa.Column("error", sa.String(), nullable=True),
+ sa.Column("revision_number", sa.String(), nullable=True),
+ sa.Column("primary_rel_path", sa.String(), nullable=True),
+ sa.Column("asf_uid", sa.String(), server_default=sa.text("('')"), nullable=False),
+ sa.Column("scheduled", sa.TIMESTAMP(), nullable=True),
+ sa.Column("inputs_hash", sa.String(), nullable=True),
+ sa.Column("project_key", sa.String(), nullable=True),
+ sa.Column("version_key", sa.String(), nullable=True),
+ sa.PrimaryKeyConstraint("id", name="pk_task"),
+ sa.ForeignKeyConstraint(["project_key"], ["project.key"], name="fk_task_project_key_project"),
+ sa.Index("ix_task_added", "added"),
+ sa.Index("ix_task_inputs_hash", "inputs_hash", unique=True),
+ sa.Index("ix_task_primary_rel_path", "primary_rel_path"),
+ sa.Index("ix_task_revision_number", "revision_number"),
+ sa.Index("ix_task_scheduled", "scheduled"),
+ sa.Index("ix_task_status", "status"),
+ sa.Index("ix_task_status_added", "status", "added"),
+ sa.Index("ix_task_version_key", "version_key"),
+ )
diff --git a/tests/unit/test_worker_defer.py b/tests/unit/test_worker_defer.py
index 3735d0dc..7c98e748 100644
--- a/tests/unit/test_worker_defer.py
+++ b/tests/unit/test_worker_defer.py
@@ -25,6 +25,12 @@ import atr.tasks.task as task
import atr.worker as worker
+def test_task_args_for_log_leaves_other_tasks_unchanged() -> None:
+ task_args = {"body": "Not a mail body"}
+
+ assert worker._task_args_for_log(sql.TaskType.MAINTENANCE, task_args) is task_args
+
+
def test_task_args_for_log_redacts_message_body_and_recipients() -> None:
task_args = {
"email_sender": "sender@apache.org",
@@ -54,12 +60,6 @@ def test_task_args_for_log_redacts_message_body_and_recipients() -> None:
assert task_args["body"] == "Secret body"
-def test_task_args_for_log_leaves_other_tasks_unchanged() -> None:
- task_args = {"body": "Not a mail body"}
-
- assert worker._task_args_for_log(sql.TaskType.MAINTENANCE, task_args) is task_args
-
-
async def test_task_process_defers_when_ldap_unavailable(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("atr.config.is_production_mode", lambda: True)
monkeypatch.setattr("atr.ldap.account_lookup", mock.AsyncMock(side_effect=ldap.UnavailableError("down")))
@@ -90,3 +90,18 @@ async def test_task_process_fails_when_handler_ldap_unavailable(monkeypatch: pyt
defer.assert_not_awaited()
result_process.assert_awaited_once_with(1, None, task.FAILED, "down")
+
+
+async def test_task_process_marks_retryable_check_failures_broken(monkeypatch: pytest.MonkeyPatch) -> None:
+ async def handler(args: object) -> None:
+ raise task.CheckRetryableError("clone failed", {"repo_url": "https://example.invalid/repo.git"})
+
+ monkeypatch.setattr("atr.config.is_production_mode", lambda: False)
+ monkeypatch.setattr("atr.config.is_ldap_configured", lambda: False)
+ monkeypatch.setattr("atr.tasks.resolve", lambda task_type: handler)
+ result_process = mock.AsyncMock()
+ monkeypatch.setattr("atr.worker._task_result_process", result_process)
+
+ await worker._task_process(1, sql.TaskType.COMPARE_SOURCE_TREES.value, {}, "alice")
+
+ result_process.assert_awaited_once_with(1, None, task.BROKEN, "clone failed")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org