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/08/25 17:22:49 UTC
(tooling-trusted-releases) 01/01: Prevent duplicate admin notifications from being stored
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 899816c8622b49c991d6eac565bf0b0961ac9330
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Tue Aug 25 18:21:18 2026 +0100
Prevent duplicate admin notifications from being stored
---
atr/models/sql.py | 35 ++++++++++++-
atr/server.py | 2 +-
atr/storage/writers/notifications.py | 14 +++---
atr/tasks/task.py | 12 +++--
migrations/versions/0122_2026.08.25_76eb8b67.py | 65 +++++++++++++++++++++++++
tests/unit/test_notifications_writer.py | 62 +++++++++++++++++++++++
tests/unit/test_task_finalise.py | 16 ++++++
7 files changed, 191 insertions(+), 15 deletions(-)
diff --git a/atr/models/sql.py b/atr/models/sql.py
index f086f149..2a812a40 100644
--- a/atr/models/sql.py
+++ b/atr/models/sql.py
@@ -25,7 +25,9 @@ import copy
import dataclasses
import datetime
import enum
+import hashlib
import ipaddress
+import json
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypeVar, assert_never, overload
import pydantic
@@ -686,8 +688,12 @@ class Notification(sqlmodel.SQLModel, table=True):
# An ATR path the notification acts on, built by the sender and never by a user
link: str | None = sqlmodel.Field(default=None)
link_text: str | None = sqlmodel.Field(default=None)
+ dedup_hash: str = sqlmodel.Field()
- __table_args__ = (sqlalchemy.Index("ix_notification_asf_uid_created", "asf_uid", "created"),)
+ __table_args__ = (
+ sqlalchemy.Index("ix_notification_asf_uid_created", "asf_uid", "created"),
+ sqlalchemy.Index("ix_notification_asf_uid_dedup_hash", "asf_uid", "dedup_hash", unique=True),
+ )
# PersonalAccessToken:
@@ -2335,6 +2341,33 @@ class WorkflowStatus(sqlmodel.SQLModel, table=True):
message: str | None = sqlmodel.Field(default=None)
+def notification_dedup_hash(level: NotificationLevel, message: str, link: str | None, link_text: str | None) -> str:
+ encoded = json.dumps([level.value, message, link, link_text])
+ return hashlib.sha3_256(encoded.encode()).hexdigest()
+
+
+def notification_insert(
+ asf_uid: str,
+ message: str,
+ level: NotificationLevel,
+ link: str | None = None,
+ link_text: str | None = None,
+) -> sqlite.Insert:
+ return (
+ sqlite.insert(Notification)
+ .values(
+ asf_uid=asf_uid,
+ created=datetime.datetime.now(datetime.UTC),
+ level=level,
+ message=message,
+ link=link,
+ link_text=link_text,
+ dedup_hash=notification_dedup_hash(level, message, link, link_text),
+ )
+ .on_conflict_do_nothing(index_elements=["asf_uid", "dedup_hash"])
+ )
+
+
def revision_key(release_key: safe.ReleaseKey | str, number: str) -> str:
return f"{release_key} {number}"
diff --git a/atr/server.py b/atr/server.py
index bceaeba6..7bf0971f 100644
--- a/atr/server.py
+++ b/atr/server.py
@@ -297,7 +297,7 @@ def _app_setup_context(app: base.QuartApp) -> None:
try:
async with storage.read(web.Committer(current_user)) as read:
rafc = read.as_foundation_committer()
- user_notifications = await rafc.notifications.pending()
+ user_notifications = await rafc.notifications.pending(limit=2)
except Exception:
log.exception("Failed to load notifications for user context")
diff --git a/atr/storage/writers/notifications.py b/atr/storage/writers/notifications.py
index 511e313d..bda64fd6 100644
--- a/atr/storage/writers/notifications.py
+++ b/atr/storage/writers/notifications.py
@@ -41,16 +41,14 @@ class FoundationCommitter:
level: sql.NotificationLevel = sql.NotificationLevel.ERROR,
link: str | None = None,
link_text: str | None = None,
- ) -> sql.Notification:
- notification = sql.Notification(
- asf_uid=self.__asf_uid,
- message=_normalised_message(message),
- level=level,
- link=link,
- link_text=link_text,
+ ) -> sql.Notification | None:
+ stmt = sql.notification_insert(self.__asf_uid, _normalised_message(message), level, link, link_text).returning(
+ sql.Notification
)
- self.__data.add(notification)
+ notification = (await self.__data.execute(stmt)).scalars().one_or_none()
await self.__data.commit()
+ if notification is None:
+ return None
self.__write_as.append_to_audit_log(
asf_uid=self.__asf_uid,
notification_id=notification.id,
diff --git a/atr/tasks/task.py b/atr/tasks/task.py
index 1fef076e..5c66ed2f 100644
--- a/atr/tasks/task.py
+++ b/atr/tasks/task.py
@@ -128,7 +128,7 @@ async def finalise_failure(
via = sql.validate_instrumented_attribute
error = ((error or "").strip()) or "unknown error"
completed = datetime.datetime.now(datetime.UTC)
- notification: sql.Notification | None = None
+ notification_id: int | None = None
async with db.ensure_session(caller_data) as data:
async with data.begin():
update_stmt = (
@@ -176,10 +176,12 @@ async def finalise_failure(
)
if (checker is None) and asf_uid and (asf_uid != constants.SYSTEM_SERVICE_UID):
message = failure_message(task_type, project_key, version_key, revision_number, primary_rel_path, error)
- notification = sql.Notification(asf_uid=asf_uid, message=message, level=sql.NotificationLevel.ERROR)
- data.add(notification)
- if notification is not None:
- storage.audit(asf_uid=notification.asf_uid, notification_id=notification.id, level=notification.level.value)
+ insert_stmt = sql.notification_insert(asf_uid, message, sql.NotificationLevel.ERROR).returning(
+ via(sql.Notification.id)
+ )
+ notification_id = (await data.execute(insert_stmt)).scalar_one_or_none()
+ if notification_id is not None:
+ storage.audit(asf_uid=asf_uid, notification_id=notification_id, level=sql.NotificationLevel.ERROR.value)
return True
diff --git a/migrations/versions/0122_2026.08.25_76eb8b67.py b/migrations/versions/0122_2026.08.25_76eb8b67.py
new file mode 100644
index 00000000..92194a7d
--- /dev/null
+++ b/migrations/versions/0122_2026.08.25_76eb8b67.py
@@ -0,0 +1,65 @@
+# 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.
+
+"""Deduplicate notifications
+
+Revision ID: 0122_2026.08.25_76eb8b67
+Revises: 0121_2026.08.21_fdda1630
+Create Date: 2026-08-25 13:46:55.901179+00:00
+"""
+
+import hashlib
+import json
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# Revision identifiers, used by Alembic
+revision: str = "0122_2026.08.25_76eb8b67"
+down_revision: str | None = "0121_2026.08.21_fdda1630"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+ with op.batch_alter_table("notification", schema=None) as batch_op:
+ batch_op.add_column(sa.Column("dedup_hash", sa.String(), nullable=True))
+
+ connection = op.get_bind()
+ rows = connection.execute(sa.text("SELECT id, level, message, link, link_text FROM notification")).fetchall()
+ for row in rows:
+ encoded = json.dumps([row.level.lower(), row.message, row.link, row.link_text])
+ connection.execute(
+ sa.text("UPDATE notification SET dedup_hash = :dedup_hash WHERE id = :id"),
+ {"dedup_hash": hashlib.sha3_256(encoded.encode()).hexdigest(), "id": row.id},
+ )
+ connection.execute(
+ sa.text(
+ "DELETE FROM notification WHERE id NOT IN (SELECT MIN(id) FROM notification GROUP BY asf_uid, dedup_hash)"
+ )
+ )
+
+ with op.batch_alter_table("notification", schema=None) as batch_op:
+ batch_op.alter_column("dedup_hash", existing_type=sa.String(), nullable=False)
+ batch_op.create_index("ix_notification_asf_uid_dedup_hash", ["asf_uid", "dedup_hash"], unique=True)
+
+
+def downgrade() -> None:
+ with op.batch_alter_table("notification", schema=None) as batch_op:
+ batch_op.drop_index("ix_notification_asf_uid_dedup_hash")
+ batch_op.drop_column("dedup_hash")
diff --git a/tests/unit/test_notifications_writer.py b/tests/unit/test_notifications_writer.py
new file mode 100644
index 00000000..cb0fb4ab
--- /dev/null
+++ b/tests/unit/test_notifications_writer.py
@@ -0,0 +1,62 @@
+# 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.
+
+import unittest.mock as mock
+from collections.abc import AsyncIterator
+
+import pytest
+import sqlalchemy
+import sqlalchemy.ext.asyncio
+import sqlmodel
+
+import atr.db as db
+import atr.models.sql as sql
+import atr.storage.writers.notifications as notifications
+
+
+@pytest.fixture
+async def sqlite_sessionmaker() -> AsyncIterator[sqlalchemy.ext.asyncio.async_sessionmaker[db.Session]]:
+ engine = sqlalchemy.ext.asyncio.create_async_engine(
+ "sqlite+aiosqlite:///:memory:",
+ connect_args={"check_same_thread": False},
+ poolclass=sqlalchemy.pool.StaticPool,
+ )
+ async with engine.begin() as conn:
+ await conn.run_sync(sqlmodel.SQLModel.metadata.create_all)
+ sessionmaker = sqlalchemy.ext.asyncio.async_sessionmaker(bind=engine, class_=db.Session, expire_on_commit=False)
+ yield sessionmaker
+ await engine.dispose()
+
+
+async def test_create_dedupes_and_audits_once(sqlite_sessionmaker) -> None:
+ async with sqlite_sessionmaker() as data:
+ write_as = mock.MagicMock()
+ writer = object.__new__(notifications.FoundationCommitter)
+ writer._FoundationCommitter__write = mock.MagicMock()
+ writer._FoundationCommitter__write_as = write_as
+ writer._FoundationCommitter__data = data
+ writer._FoundationCommitter__asf_uid = "alice"
+
+ first = await writer.create("The SVN dist repository was not reachable")
+ second = await writer.create("The SVN dist repository was not reachable")
+
+ assert first is not None
+ assert second is None
+ notification = (await data.execute(sqlmodel.select(sql.Notification))).scalar_one()
+ assert notification.id == first.id
+ assert notification.message == "The SVN dist repository was not reachable"
+ write_as.append_to_audit_log.assert_called_once_with(asf_uid="alice", notification_id=first.id, level="error")
diff --git a/tests/unit/test_task_finalise.py b/tests/unit/test_task_finalise.py
index 06e337d2..6eb72da4 100644
--- a/tests/unit/test_task_finalise.py
+++ b/tests/unit/test_task_finalise.py
@@ -59,6 +59,22 @@ async def test_finalise_failure_notifies_for_non_check_tasks(sqlite_sessionmaker
assert "failed" in notification.message
+async def test_finalise_failure_dedupes_repeated_notifications(sqlite_sessionmaker) -> None:
+ async with sqlite_sessionmaker() as data:
+ first = _active_task(pid=1234, task_type=sql.TaskType.SBOM_GENERATE)
+ second = _active_task(pid=1234, task_type=sql.TaskType.SBOM_GENERATE)
+ second.inputs_hash = "blake3:4567"
+ data.add(first)
+ data.add(second)
+ await data.commit()
+
+ assert await task.finalise_failure(first.id, 1234, "boom", task.FAILED, caller_data=data) is True
+ assert await task.finalise_failure(second.id, 1234, "boom", task.FAILED, caller_data=data) is True
+
+ notification = (await data.execute(sqlmodel.select(sql.Notification))).scalar_one()
+ assert notification.asf_uid == "alice"
+
+
async def test_finalise_failure_skips_notification_for_the_system_uid(sqlite_sessionmaker) -> None:
async with sqlite_sessionmaker() as data:
task_row = _active_task(pid=1234, asf_uid="system", task_type=sql.TaskType.SBOM_GENERATE)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org