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 2025/02/21 18:11:08 UTC
(tooling-atr-experiments) branch main updated: Add a task manager and use tasks for package verification
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tooling-atr-experiments.git
The following commit(s) were added to refs/heads/main by this push:
new e7582a3 Add a task manager and use tasks for package verification
e7582a3 is described below
commit e7582a304d9b8dfd2172119f49b65639edf6f23b
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Fri Feb 21 20:10:44 2025 +0200
Add a task manager and use tasks for package verification
---
atr/blueprints/secret/secret.py | 31 --
atr/db/models.py | 24 +-
atr/manager.py | 355 +++++++++++++++++++++
atr/routes.py | 182 ++++++-----
atr/server.py | 12 +
atr/templates/candidate-attach.html | 0
atr/templates/candidate-review.html | 44 ++-
atr/templates/package-check.html | 145 +++++++++
atr/templates/secret/tasks-add-random.html | 42 ---
atr/worker.py | 197 ++++++++++--
docs/plan.html | 36 ++-
docs/plan.md | 22 +-
...al_schema.py => 861b538fac8c_initial_schema.py} | 6 +-
13 files changed, 895 insertions(+), 201 deletions(-)
diff --git a/atr/blueprints/secret/secret.py b/atr/blueprints/secret/secret.py
index 1d8a2e7..875126e 100644
--- a/atr/blueprints/secret/secret.py
+++ b/atr/blueprints/secret/secret.py
@@ -15,9 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-import datetime
import json
-import secrets
import httpx
from quart import current_app, flash, redirect, render_template, request, url_for
@@ -37,7 +35,6 @@ from atr.db.models import (
PublicSigningKey,
Release,
Task,
- TaskStatus,
VotePolicy,
)
from atr.db.service import get_pmcs
@@ -241,31 +238,3 @@ async def secret_keys_delete_all() -> str:
await db_session.delete(key)
return f"Deleted {count} keys"
-
-
-@blueprint.route("/tasks/add-random", methods=["GET", "POST"])
-async def secret_tasks_add_random() -> str | Response:
- """Add a random task to the queue for testing."""
- if request.method == "POST":
- async with get_session() as db_session:
- async with db_session.begin():
- # Create a random task
- task = Task(
- id=None,
- task_type="example",
- task_args=json.dumps(
- {
- "random_number": secrets.randbelow(100),
- "timestamp": datetime.datetime.now(datetime.UTC).isoformat(),
- }
- ),
- status=TaskStatus.QUEUED,
- )
- db_session.add(task)
- # Flush to get the task ID
- await db_session.flush()
- await flash(f"Added random task (ID: {task.id})", "success")
-
- return redirect(url_for("secret_blueprint.secret_tasks_add_random"))
-
- return await render_template("secret/tasks-add-random.html")
diff --git a/atr/db/models.py b/atr/db/models.py
index ce37d13..6c9999e 100644
--- a/atr/db/models.py
+++ b/atr/db/models.py
@@ -19,7 +19,7 @@
import datetime
from enum import Enum
-from typing import Optional
+from typing import Any, Optional
from pydantic import BaseModel
from sqlalchemy import JSON, CheckConstraint, Column, Index
@@ -163,6 +163,11 @@ class Package(SQLModel, table=True):
release_key: str | None = Field(default=None, foreign_key="release.storage_key")
release: Optional["Release"] = Relationship(back_populates="packages")
+ # One-to-many: A package can have multiple tasks
+ tasks: list["Task"] = Relationship(
+ back_populates="package", sa_relationship_kwargs={"cascade": "all, delete-orphan"}
+ )
+
class VoteEntry(BaseModel):
result: bool
@@ -208,15 +213,20 @@ class Task(SQLModel, table=True):
"""A task in the task queue."""
id: int | None = Field(default=None, primary_key=True)
- task_type: str
- task_args: str = Field(sa_column=Column(JSON))
status: TaskStatus = Field(default=TaskStatus.QUEUED, index=True)
+ task_type: str
+ task_args: Any = Field(sa_column=Column(JSON))
added: datetime.datetime = Field(default_factory=lambda: datetime.datetime.now(datetime.UTC), index=True)
started: datetime.datetime | None = None
- completed: datetime.datetime | None = None
pid: int | None = None
+ completed: datetime.datetime | None = None
+ result: Any | None = Field(default=None, sa_column=Column(JSON))
error: str | None = None
+ # Package relationship
+ package_sha3: str | None = Field(default=None, foreign_key="package.artifact_sha3")
+ package: Package | None = Relationship(back_populates="tasks")
+
# Create an index on status and added for efficient task claiming
__table_args__ = (
Index("ix_task_status_added", "status", "added"),
@@ -231,9 +241,9 @@ class Task(SQLModel, table=True):
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
- OR (status = 'COMPLETED' AND completed IS NOT NULL)
- -- ACTIVE -> FAILED requires setting completed time and error
+ -- 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)
)
""",
diff --git a/atr/manager.py b/atr/manager.py
new file mode 100644
index 0000000..7436007
--- /dev/null
+++ b/atr/manager.py
@@ -0,0 +1,355 @@
+# 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.
+
+"""Worker process manager."""
+
+import asyncio
+import logging
+import os
+import signal
+import sys
+from datetime import UTC, datetime
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from atr.db import get_session
+
+# Configure logging
+logging.basicConfig(
+ format="[%(asctime)s.%(msecs)03d] [%(levelname)s] %(message)s",
+ level=logging.INFO,
+ datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger(__name__)
+
+
+class WorkerProcess:
+ """Interface to control a worker process."""
+
+ def __init__(self, process: asyncio.subprocess.Process, started: datetime):
+ self.process = process
+ self.started = started
+ self.last_checked = started
+
+ @property
+ def pid(self) -> int | None:
+ return self.process.pid
+
+ async def is_running(self) -> bool:
+ """Check if the process is still running."""
+ if self.process.returncode is not None:
+ # Process has already exited
+ return False
+
+ if not self.pid:
+ # Process did not start
+ return False
+
+ try:
+ os.kill(self.pid, 0)
+ self.last_checked = datetime.now(UTC)
+ return True
+ except ProcessLookupError:
+ # Process no longer exists
+ return False
+ except PermissionError:
+ # Process exists but we don't have permission to signal it
+ # This shouldn't happen in our case since we own the process
+ logger.warning(f"Permission error checking process {self.pid}")
+ return False
+
+
+class WorkerManager:
+ """Manager for a pool of worker processes."""
+
+ def __init__(
+ self,
+ min_workers: int = 2,
+ max_workers: int = 4,
+ check_interval: float = 5.0,
+ max_task_seconds: float = 300.0,
+ ):
+ self.min_workers = min_workers
+ self.max_workers = max_workers
+ self.check_interval = check_interval
+ self.max_task_seconds = max_task_seconds
+ self.workers: dict[int, WorkerProcess] = {}
+ self.running = False
+ self.check_task: asyncio.Task | None = None
+
+ async def start(self) -> None:
+ """Start the worker manager."""
+ if self.running:
+ return
+
+ self.running = True
+ logger.info("Starting worker manager in %s", os.getcwd())
+
+ # Start initial workers
+ for _ in range(self.min_workers):
+ await self.spawn_worker()
+
+ # Start monitoring task
+ self.check_task = asyncio.create_task(self.monitor_workers())
+
+ async def stop(self) -> None:
+ """Stop all workers and the manager."""
+ if not self.running:
+ return
+
+ self.running = False
+ logger.info("Stopping worker manager")
+
+ # Cancel monitoring task
+ if self.check_task:
+ self.check_task.cancel()
+ try:
+ await self.check_task
+ except asyncio.CancelledError:
+ ...
+
+ # Stop all workers
+ await self.stop_all_workers()
+
+ async def stop_all_workers(self) -> None:
+ """Stop all worker processes."""
+ for worker in list(self.workers.values()):
+ if worker.pid:
+ try:
+ os.kill(worker.pid, signal.SIGTERM)
+ except ProcessLookupError:
+ # The process may have already exited
+ ...
+ except Exception as e:
+ logger.error(f"Error stopping worker {worker.pid}: {e}")
+
+ # Wait for processes to exit
+ for worker in list(self.workers.values()):
+ try:
+ await asyncio.wait_for(worker.process.wait(), timeout=5.0)
+ except TimeoutError:
+ if worker.pid:
+ try:
+ os.kill(worker.pid, signal.SIGKILL)
+ except ProcessLookupError:
+ # The process may have already exited
+ ...
+ except Exception as e:
+ logger.error(f"Error force killing worker {worker.pid}: {e}")
+
+ self.workers.clear()
+
+ async def spawn_worker(self) -> None:
+ """Spawn a new worker process."""
+ if len(self.workers) >= self.max_workers:
+ return
+
+ try:
+ # Get the absolute path to the project root (i.e. atr/..)
+ project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+ # Ensure PYTHONPATH includes our project root
+ env = os.environ.copy()
+ python_path = env.get("PYTHONPATH", "")
+ env["PYTHONPATH"] = f"{project_root}:{python_path}" if python_path else project_root
+
+ # Get absolute path to worker script
+ worker_script = os.path.join(project_root, "atr", "worker.py")
+
+ # Start worker process with the updated environment
+ # Use preexec_fn to create new process group
+ process = await asyncio.create_subprocess_exec(
+ sys.executable,
+ worker_script,
+ stdout=asyncio.subprocess.DEVNULL,
+ stderr=asyncio.subprocess.DEVNULL,
+ env=env,
+ preexec_fn=os.setsid,
+ )
+
+ worker = WorkerProcess(process, datetime.now(UTC))
+ if worker.pid:
+ self.workers[worker.pid] = worker
+ logger.info(f"Started worker process {worker.pid}")
+ else:
+ # TODO: We should count failures
+ # We could perhaps stop the manager over a certain count
+ logger.error("Failed to start worker process")
+ except Exception as e:
+ logger.error(f"Error spawning worker: {e}")
+
+ async def monitor_workers(self) -> None:
+ """Monitor worker processes and restart them if needed."""
+ while self.running:
+ try:
+ await self.check_workers()
+ await asyncio.sleep(self.check_interval)
+ except asyncio.CancelledError:
+ break
+ except Exception as e:
+ logger.error(f"Error in worker monitor: {e}")
+ # TODO: How long should we wait before trying again?
+ await asyncio.sleep(1.0)
+
+ async def check_workers(self) -> None:
+ """Check worker processes and restart if needed."""
+ exited_workers = []
+
+ # Check each worker first
+ for pid, worker in list(self.workers.items()):
+ # Check if process is running
+ if not await worker.is_running():
+ exited_workers.append(pid)
+ logger.info(f"Worker {pid} has exited")
+ continue
+
+ # Check if worker has been processing its task for too long
+ # This also stops tasks if they have indeed been running for too long
+ if await self.check_task_duration(pid, worker):
+ exited_workers.append(pid)
+
+ # Remove exited workers
+ for pid in exited_workers:
+ self.workers.pop(pid, None)
+
+ # # Check for active tasks
+ # try:
+ # async with get_session() as session:
+ # result = await session.execute(
+ # text("""
+ # SELECT COUNT(*)
+ # FROM task
+ # WHERE status = 'QUEUED'
+ # """)
+ # )
+ # queued_count = result.scalar()
+ # logger.info(f"Found {queued_count} queued tasks waiting for workers")
+ # except Exception as e:
+ # logger.error(f"Error checking queued tasks: {e}")
+
+ # Spawn new workers if needed
+ await self.maintain_worker_pool()
+
+ # Reset any tasks that were being processed by exited workers
+ if exited_workers:
+ await self.reset_broken_tasks(exited_workers)
+
+ async def terminate_long_running_task(
+ self, session: AsyncSession, worker: WorkerProcess, task_id: int, pid: int
+ ) -> None:
+ """
+ Terminate a task that has been running for too long.
+ Updates the task status and terminates the worker process.
+ """
+ try:
+ # Mark the task as failed
+ await session.execute(
+ text("""
+ UPDATE task
+ SET status = 'FAILED', completed = :now, error = :error
+ WHERE id = :task_id
+ AND status = 'ACTIVE'
+ """),
+ {
+ "now": datetime.now(UTC),
+ "task_id": task_id,
+ "error": f"Task terminated after exceeding time limit of {self.max_task_seconds} seconds",
+ },
+ )
+ if worker.pid:
+ os.kill(worker.pid, signal.SIGTERM)
+ logger.info(f"Worker {pid} terminated after processing task {task_id} for > {self.max_task_seconds}s")
+ except ProcessLookupError:
+ return
+ except Exception as e:
+ logger.error(f"Error stopping long-running worker {pid}: {e}")
+
+ async def check_task_duration(self, pid: int, worker: WorkerProcess) -> bool:
+ """
+ Check if a worker has been processing its task for too long.
+ Returns True if the worker has been terminated.
+ """
+ try:
+ async with get_session() as session:
+ async with session.begin():
+ result = await session.execute(
+ text("""
+ SELECT id, started FROM task
+ WHERE status = 'ACTIVE'
+ AND pid = :pid
+ """),
+ {"pid": pid},
+ )
+ task = result.first()
+ if not task or not task[1]:
+ return False
+
+ task_id, started = task
+ task_duration = (datetime.now(UTC) - started).total_seconds()
+ if task_duration > self.max_task_seconds:
+ await self.terminate_long_running_task(session, worker, task_id, pid)
+ return True
+
+ return False
+ except Exception as e:
+ logger.error(f"Error checking task duration for worker {pid}: {e}")
+ # TODO: Return True? False? Maybe None would be more suitable, or propagate the error
+ return True
+
+ async def maintain_worker_pool(self) -> None:
+ """Ensure we maintain the minimum number of workers."""
+ current_count = len(self.workers)
+ if current_count < self.min_workers:
+ logger.info(f"Worker pool below minimum ({current_count} < {self.min_workers}), spawning new workers")
+ while len(self.workers) < self.min_workers:
+ await self.spawn_worker()
+ logger.info(f"Worker pool restored to {len(self.workers)} workers")
+
+ async def reset_broken_tasks(self, exited_pids: list[int]) -> None:
+ """Reset any tasks that were being processed by exited workers."""
+ try:
+ async with get_session() as session:
+ async with session.begin():
+ # Generate named parameters for each PID
+ placeholders = ",".join(f":pid_{i}" for i in range(len(exited_pids)))
+ params = {f"pid_{i}": pid for i, pid in enumerate(exited_pids)}
+
+ # Execute update with proper parameter binding
+ await session.execute(
+ text(f"""
+ UPDATE task
+ SET status = 'QUEUED', started = NULL, pid = NULL
+ WHERE status = 'ACTIVE'
+ AND pid IN ({placeholders})
+ """),
+ params,
+ )
+ except Exception as e:
+ logger.error(f"Error resetting broken tasks: {e}")
+
+
+# Global worker manager instance
+worker_manager: WorkerManager | None = None
+
+
+def get_worker_manager() -> WorkerManager:
+ """Get the global worker manager instance."""
+ global worker_manager
+ if worker_manager is None:
+ worker_manager = WorkerManager()
+ return worker_manager
diff --git a/atr/routes.py b/atr/routes.py
index 869f954..272665f 100644
--- a/atr/routes.py
+++ b/atr/routes.py
@@ -27,7 +27,7 @@ import tempfile
from collections.abc import Sequence
from contextlib import asynccontextmanager
from pathlib import Path
-from typing import Any, BinaryIO, cast
+from typing import cast
import aiofiles
import aiofiles.os
@@ -54,6 +54,8 @@ from atr.db.models import (
Release,
ReleasePhase,
ReleaseStage,
+ Task,
+ TaskStatus,
)
from .db import get_session
@@ -410,6 +412,7 @@ async def root_candidate_create() -> Response | str:
@require(Requirements.committer)
async def root_candidate_signatures_verify(release_key: str) -> str:
"""Verify the signatures for all packages in a release candidate."""
+ # TODO: This entire endpoint can be removed
session = await session_read()
if session is None:
raise ASFQuartException("Not authenticated", errorcode=401)
@@ -424,7 +427,7 @@ async def root_candidate_signatures_verify(release_key: str) -> str:
# For now, for debugging, we'll just get all keys in the database
statement = select(PublicSigningKey)
- all_signing_keys = (await db_session.execute(statement)).scalars().all()
+ # all_signing_keys = (await db_session.execute(statement)).scalars().all()
statement = (
select(Release)
@@ -436,7 +439,7 @@ async def root_candidate_signatures_verify(release_key: str) -> str:
raise ASFQuartException("Release not found", errorcode=404)
# Get all ASCII-armored keys associated with the PMC
- ascii_armored_keys = [key.ascii_armored_key for key in all_signing_keys]
+ # ascii_armored_keys = [key.ascii_armored_key for key in all_signing_keys]
# Verify each package's signature
verification_results = []
@@ -459,7 +462,8 @@ async def root_candidate_signatures_verify(release_key: str) -> str:
result["error"] = "Package signature file not found"
else:
# Verify the signature
- result = await verify_gpg_signature(artifact_path, signature_path, ascii_armored_keys)
+ result = {}
+ # await verify_gpg_signature(artifact_path, signature_path, ascii_armored_keys)
result["file"] = package.artifact_sha3
result["filename"] = package.filename
@@ -701,10 +705,13 @@ async def root_candidate_review() -> str:
# TODO: This duplicates code in root_package_add
release_pmc = selectinload(cast(InstrumentedAttribute[PMC], Release.pmc))
release_packages = selectinload(cast(InstrumentedAttribute[list[Package]], Release.packages))
+ package_tasks = selectinload(cast(InstrumentedAttribute[list[Package]], Release.packages)).selectinload(
+ cast(InstrumentedAttribute[list[Task]], Package.tasks)
+ )
release_product_line = selectinload(cast(InstrumentedAttribute[ProductLine], Release.product_line))
statement = (
select(Release)
- .options(release_pmc, release_packages, release_product_line)
+ .options(release_pmc, release_packages, package_tasks, release_product_line)
.join(PMC)
.where(Release.stage == ReleaseStage.CANDIDATE)
)
@@ -740,7 +747,7 @@ async def delete_package_files(package: Package, uploads_path: Path) -> None:
await aiofiles.os.remove(signature_path)
-async def validate_package_deletion(
+async def package_from_data(
db_session: AsyncSession, artifact_sha3: str, release_key: str, session_uid: str
) -> Package:
"""Validate package deletion request and return the package if valid."""
@@ -769,7 +776,7 @@ async def validate_package_deletion(
# Check permissions
if package.release and package.release.pmc:
if session_uid not in package.release.pmc.pmc_members and session_uid not in package.release.pmc.committers:
- raise FlashError("You don't have permission to delete this package")
+ raise FlashError("You don't have permission to access this package")
return package
@@ -793,7 +800,7 @@ async def root_package_delete() -> Response:
async with get_session() as db_session:
async with db_session.begin():
try:
- package = await validate_package_deletion(db_session, artifact_sha3, release_key, session.uid)
+ package = await package_from_data(db_session, artifact_sha3, release_key, session.uid)
await delete_package_files(package, Path(get_release_storage_dir()))
await db_session.delete(package)
except FlashError as e:
@@ -943,73 +950,6 @@ async def user_keys_add_session(
}
-async def verify_gpg_signature(artifact_path: Path, signature_path: Path, public_keys: list[str]) -> dict[str, Any]:
- """
- Verify a GPG signature for a release artifact.
- Returns a dictionary with verification results and debug information.
- """
- try:
- with open(signature_path, "rb") as sig_file:
- return await verify_gpg_signature_file(sig_file, artifact_path, public_keys)
- except Exception as e:
- return {
- "verified": False,
- "error": str(e),
- "status": "Verification failed",
- "debug_info": {"exception_type": type(e).__name__, "exception_message": str(e)},
- }
-
-
-async def verify_gpg_signature_file(
- sig_file: BinaryIO, artifact_path: Path, ascii_armored_keys: list[str]
-) -> dict[str, Any]:
- """Verify a GPG signature for a file."""
- async with ephemeral_gpg_home() as gpg_home:
- gpg = gnupg.GPG(gnupghome=gpg_home)
-
- # Import all PMC public signing keys
- for key in ascii_armored_keys:
- import_result = await asyncio.to_thread(gpg.import_keys, key)
- if not import_result.fingerprints:
- # TODO: Log warning about invalid key?
- continue
- verified = await asyncio.to_thread(gpg.verify_file, sig_file, str(artifact_path))
-
- # Collect all available information for debugging
- debug_info = {
- "key_id": verified.key_id or "Not available",
- "fingerprint": verified.fingerprint.lower() if verified.fingerprint else "Not available",
- "pubkey_fingerprint": verified.pubkey_fingerprint.lower() if verified.pubkey_fingerprint else "Not available",
- "creation_date": verified.creation_date or "Not available",
- "timestamp": verified.timestamp or "Not available",
- "username": verified.username or "Not available",
- "status": verified.status or "Not available",
- "valid": bool(verified),
- "trust_level": verified.trust_level if hasattr(verified, "trust_level") else "Not available",
- "trust_text": verified.trust_text if hasattr(verified, "trust_text") else "Not available",
- "stderr": verified.stderr if hasattr(verified, "stderr") else "Not available",
- "num_pmc_keys": len(ascii_armored_keys),
- }
-
- if not verified:
- return {
- "verified": False,
- "error": "No valid signature found",
- "status": "Invalid signature",
- "debug_info": debug_info,
- }
-
- return {
- "verified": True,
- "key_id": verified.key_id,
- "timestamp": verified.timestamp,
- "username": verified.username or "Unknown",
- "email": verified.pubkey_fingerprint.lower() or "Unknown",
- "status": "Valid signature",
- "debug_info": debug_info,
- }
-
-
async def validate_release_deletion(db_session: AsyncSession, release_key: str, session_uid: str) -> Release:
"""Validate release deletion request and return the release if valid."""
if Release.pmc is None:
@@ -1070,3 +1010,95 @@ async def root_release_delete() -> Response:
await flash("Release deleted successfully", "success")
return redirect(url_for("root_candidate_review"))
+
+
+async def get_package_tasks_status(db_session: AsyncSession, artifact_sha3: str) -> tuple[Sequence[Task], bool]:
+ """
+ Get all tasks for a package and determine if any are still in progress.
+ Returns tuple[Sequence[Task], bool]: List of tasks and whether any are still in progress
+ TODO: Could instead give active count and total count
+ """
+ statement = select(Task).where(Task.package_sha3 == artifact_sha3)
+ tasks = (await db_session.execute(statement)).scalars().all()
+ has_active_tasks = any(task.status in [TaskStatus.QUEUED, TaskStatus.ACTIVE] for task in tasks)
+ return tasks, has_active_tasks
+
+
+@APP.route("/package/check", methods=["GET", "POST"])
+@require(Requirements.committer)
+async def root_package_check() -> str | Response:
+ """Show or create package verification tasks."""
+ session = await session_read()
+ if (session is None) or (session.uid is None):
+ raise ASFQuartException("Not authenticated", errorcode=401)
+
+ # Get parameters from either form data (POST) or query args (GET)
+ if request.method == "POST":
+ form = await request.form
+ artifact_sha3 = form.get("artifact_sha3")
+ release_key = form.get("release_key")
+ else:
+ artifact_sha3 = request.args.get("artifact_sha3")
+ release_key = request.args.get("release_key")
+
+ if not artifact_sha3 or not release_key:
+ await flash("Missing required parameters", "error")
+ return redirect(url_for("root_candidate_review"))
+
+ async with get_session() as db_session:
+ async with db_session.begin():
+ # Get the package and verify permissions
+ try:
+ package = await package_from_data(db_session, artifact_sha3, release_key, session.uid)
+ except FlashError as e:
+ await flash(str(e), "error")
+ return redirect(url_for("root_candidate_review"))
+
+ if request.method == "POST":
+ # Check if package already has active tasks
+ tasks, has_active_tasks = await get_package_tasks_status(db_session, artifact_sha3)
+ if has_active_tasks:
+ await flash("Package verification is already in progress", "warning")
+ return redirect(url_for("root_candidate_review"))
+
+ if not package.release or not package.release.pmc:
+ await flash("Could not determine PMC for package", "error")
+ return redirect(url_for("root_candidate_review"))
+
+ if package.signature_sha3 is None:
+ await flash("Package has no signature", "error")
+ return redirect(url_for("root_candidate_review"))
+
+ # Create verification tasks
+ tasks = [
+ Task(
+ status=TaskStatus.QUEUED,
+ task_type="verify_archive_integrity",
+ task_args=["releases/" + artifact_sha3],
+ package_sha3=artifact_sha3,
+ ),
+ Task(
+ status=TaskStatus.QUEUED,
+ task_type="verify_signature",
+ task_args=[
+ package.release.pmc.project_name,
+ "releases/" + artifact_sha3,
+ "releases/" + package.signature_sha3,
+ ],
+ package_sha3=artifact_sha3,
+ ),
+ ]
+ for task in tasks:
+ db_session.add(task)
+
+ await flash(f"Added verification tasks for package {package.filename}", "success")
+ return redirect(url_for("root_package_check", artifact_sha3=artifact_sha3, release_key=release_key))
+ else:
+ # Get all tasks for this package for GET request
+ tasks, _ = await get_package_tasks_status(db_session, artifact_sha3)
+ return await render_template(
+ "package-check.html",
+ package=package,
+ tasks=tasks,
+ format_file_size=format_file_size,
+ )
diff --git a/atr/server.py b/atr/server.py
index 51ce772..dbf0cf5 100644
--- a/atr/server.py
+++ b/atr/server.py
@@ -32,6 +32,7 @@ from asfquart.base import QuartApp
from atr.blueprints import register_blueprints
from atr.config import AppConfig, config_dict
from atr.db import create_database
+from atr.manager import get_worker_manager
# WARNING: Don't run with debug turned on in production!
DEBUG: bool = config("DEBUG", default=True, cast=bool)
@@ -104,8 +105,19 @@ def create_app(app_config: type[AppConfig]) -> QuartApp:
"is_admin": is_admin,
}
+ @app.before_serving
+ async def startup() -> None:
+ """Start services before the app starts serving requests."""
+ # Start the worker manager
+ worker_manager = get_worker_manager()
+ await worker_manager.start()
+
@app.after_serving
async def shutdown() -> None:
+ """Clean up services after the app stops serving requests."""
+ # Stop the worker manager
+ worker_manager = get_worker_manager()
+ await worker_manager.stop()
app.background_tasks.clear()
# Configure logging
diff --git a/atr/templates/candidate-attach.html b/atr/templates/candidate-attach.html
new file mode 100644
index 0000000..e69de29
diff --git a/atr/templates/candidate-review.html b/atr/templates/candidate-review.html
index 19648a6..bc3a6d3 100644
--- a/atr/templates/candidate-review.html
+++ b/atr/templates/candidate-review.html
@@ -61,10 +61,14 @@
color: white;
}
- .delete-form {
+ .button-form {
display: inline-block;
}
+ .button-form+.button-form {
+ margin-left: 1rem;
+ }
+
.delete-button {
padding: 0.5rem 1rem;
background: #dc3545;
@@ -148,15 +152,11 @@
<span class="candidate-meta-item">Created: {{ release.created.strftime("%Y-%m-%d %H:%M UTC") }}</span>
</div>
<div class="candidate-actions">
- <a class="verify-link"
- href="{{ url_for('root_candidate_signatures_verify', release_key=release.storage_key) }}">
- Verify signatures
- </a>
<a class="verify-link"
href="{{ url_for('root_package_add', storage_key=release.storage_key) }}">Add package</a>
<form method="post"
action="{{ url_for('root_release_delete') }}"
- class="delete-form">
+ class="button-form">
<input type="hidden" name="release_key" value="{{ release.storage_key }}" />
<button type="submit"
class="delete-button"
@@ -204,12 +204,42 @@
<th>Uploaded</th>
<td>{{ package.uploaded.strftime("%Y-%m-%d %H:%M UTC") }}</td>
</tr>
+ <tr>
+ <th>Package checks</th>
+ <td>
+ {% set total_tasks = package.tasks|length %}
+ {% set completed_tasks = package.tasks|selectattr("status", "in", ["COMPLETED", "FAILED"])|list|length %}
+ {% if total_tasks == 0 %}
+ Not started
+ {% else %}
+ {% if completed_tasks == total_tasks %}
+ Completed ({{ completed_tasks }}/{{ total_tasks }})
+ {% else %}
+ Ongoing ({{ completed_tasks }}/{{ total_tasks }})
+ {% endif %}
+ {% endif %}
+ </td>
+ </tr>
<tr>
<th>Actions</th>
<td class="prose">
+ {% if package.tasks %}
+ <a href="{{ url_for('root_package_check', artifact_sha3=package.artifact_sha3, release_key=release.storage_key) }}"
+ class="verify-link">Show check status</a>
+ {% else %}
+ <form method="post"
+ action="{{ url_for('root_package_check') }}"
+ class="button-form">
+ <input type="hidden"
+ name="artifact_sha3"
+ value="{{ package.artifact_sha3 }}" />
+ <input type="hidden" name="release_key" value="{{ release.storage_key }}" />
+ <button type="submit" class="verify-link">Check package</button>
+ </form>
+ {% endif %}
<form method="post"
action="{{ url_for('root_package_delete') }}"
- class="delete-form">
+ class="button-form">
<input type="hidden"
name="artifact_sha3"
value="{{ package.artifact_sha3 }}" />
diff --git a/atr/templates/package-check.html b/atr/templates/package-check.html
new file mode 100644
index 0000000..a80bbaf
--- /dev/null
+++ b/atr/templates/package-check.html
@@ -0,0 +1,145 @@
+{% extends "layouts/base.html" %}
+
+{% block title %}
+ Package check status ~ ATR
+{% endblock title %}
+
+{% block description %}
+ View the status and results of package verification tasks.
+{% endblock description %}
+
+{% block stylesheets %}
+ {{ super() }}
+ <style>
+ .task-list {
+ margin: 1rem 0;
+ }
+
+ .task-item {
+ border: 1px solid #ddd;
+ border-radius: 4px;
+ padding: 1rem;
+ margin-bottom: 1rem;
+ }
+
+ .task-header {
+ display: flex;
+ justify-content: space-between;
+ align-items: center;
+ margin-bottom: 0.5rem;
+ }
+
+ .task-type {
+ font-weight: 600;
+ }
+
+ .task-status {
+ padding: 0.25rem 0.5rem;
+ border-radius: 4px;
+ font-size: 0.9em;
+ }
+
+ .status-queued {
+ background: #f8f9fa;
+ border: 1px solid #dee2e6;
+ }
+
+ .status-active {
+ background: #cff4fc;
+ border: 1px solid #9eeaf9;
+ }
+
+ .status-completed {
+ background: #d1e7dd;
+ border: 1px solid #a3cfbb;
+ }
+
+ .status-failed {
+ background: #f8d7da;
+ border: 1px solid #f5c6cb;
+ }
+
+ .task-details {
+ margin-top: 0.5rem;
+ font-size: 0.9em;
+ }
+
+ .task-result {
+ margin-top: 0.5rem;
+ padding: 0.5rem;
+ background: #f8f9fa;
+ border-radius: 4px;
+ white-space: pre-wrap;
+ }
+
+ .package-info {
+ margin-bottom: 2rem;
+ padding: 1rem;
+ background: #f8f9fa;
+ border: 1px solid #dee2e6;
+ border-radius: 4px;
+ }
+
+ .package-info h2 {
+ margin-top: 0;
+ }
+
+ .back-link {
+ display: inline-block;
+ margin-bottom: 1rem;
+ color: #0d6efd;
+ text-decoration: none;
+ }
+
+ .back-link:hover {
+ text-decoration: underline;
+ }
+ </style>
+{% endblock stylesheets %}
+
+{% block content %}
+ <a href="{{ url_for('root_candidate_review') }}" class="back-link">← Back to Release Candidates</a>
+
+ <div class="package-info">
+ <h2>Package details</h2>
+ <p>
+ <strong>Filename:</strong> {{ package.filename }}
+ </p>
+ <p>
+ <strong>Type:</strong> {{ package.artifact_type }}
+ </p>
+ <p>
+ <strong>Size:</strong> {{ format_file_size(package.bytes_size) }}
+ </p>
+ <p>
+ <strong>Uploaded:</strong> {{ package.uploaded.strftime("%Y-%m-%d %H:%M UTC") }}
+ </p>
+ </div>
+
+ <p>
+ <strong>Note:</strong> Refresh this page to see the latest status of the verification tasks.
+ </p>
+
+ <h2>Verification tasks</h2>
+
+ <div class="task-list">
+ {% if tasks %}
+ {% for task in tasks %}
+ <div class="task-item">
+ <div class="task-header">
+ <span class="task-type">{{ task.task_type }}</span>
+ <span class="task-status status-{{ task.status.value.lower() }}">{{ task.status.value }}</span>
+ </div>
+ <div class="task-details">
+ <div>Started: {{ task.started.strftime("%Y-%m-%d %H:%M UTC") if task.started else "Not started" }}</div>
+ <div>Completed: {{ task.completed.strftime("%Y-%m-%d %H:%M UTC") if task.completed else "Not completed" }}</div>
+ {% if task.result %}<div class="task-result">{{ task.result }}</div>{% endif %}
+ {% if task.error %}<div class="task-result">Error: {{ task.error }}</div>{% endif %}
+ </div>
+ </div>
+ {% endfor %}
+ {% else %}
+ <p>No verification tasks found for this package.</p>
+ {% endif %}
+ </div>
+{% endblock content %}
diff --git a/atr/templates/secret/tasks-add-random.html b/atr/templates/secret/tasks-add-random.html
deleted file mode 100644
index b83c8a1..0000000
--- a/atr/templates/secret/tasks-add-random.html
+++ /dev/null
@@ -1,42 +0,0 @@
-{% extends "layouts/base.html" %}
-
-{% block title %}
- Add random task ~ ATR
-{% endblock title %}
-
-{% block description %}
- Add a random task to the queue for testing.
-{% endblock description %}
-
-{% block stylesheets %}
- {{ super() }}
- <style>
- .status-message {
- margin: 1.5rem 0;
- padding: 1rem;
- border-radius: 4px;
- }
-
- .status-message.success {
- background: #d4edda;
- border: 1px solid #c3e6cb;
- color: #155724;
- }
-
- .status-message.error {
- background: #f8d7da;
- border: 1px solid #f5c6cb;
- color: #721c24;
- }
- </style>
-{% endblock stylesheets %}
-
-{% block content %}
- <div class="container mt-4">
- <h1>Add random task</h1>
-
- <form method="post" class="mt-4">
- <button type="submit" class="btn btn-primary">Add random task</button>
- </form>
- </div>
-{% endblock content %}
diff --git a/atr/worker.py b/atr/worker.py
index 7c4d3e2..6aca90b 100644
--- a/atr/worker.py
+++ b/atr/worker.py
@@ -23,17 +23,27 @@
# need to check wall clock time as well as CPU time.
import datetime
+import json
import logging
import os
import resource
+import shutil
import signal
import sys
+import tarfile
+import tempfile
import time
+from collections.abc import Generator
+from contextlib import contextmanager
from datetime import UTC
-from typing import NoReturn
+from typing import Any, BinaryIO, cast
+import gnupg
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session
+from sqlalchemy.sql import select
+
+from atr.db.models import PMC, PMCKeyLink, PublicSigningKey
# Configure logging
logging.basicConfig(
@@ -47,15 +57,112 @@ logger = logging.getLogger(__name__)
CPU_LIMIT_SECONDS = 300
MEMORY_LIMIT_BYTES = 1024 * 1024 * 1024
-# Create database engine
-engine = create_engine("sqlite:///atr.db", echo=False)
-
# # Create tables if they don't exist
# SQLModel.metadata.create_all(engine)
+class VerifyError(Exception):
+ """Error during verification."""
+
+ def __init__(self, message: str, *result: Any) -> None:
+ self.message = message
+ self.result = tuple(result)
+
+
+def verify_archive_integrity(path: str, chunk_size: int = 4096) -> int:
+ """Verify a .tar.gz file and compute its uncompressed size."""
+ total_size = 0
+
+ with tarfile.open(path, mode="r|gz") as tf:
+ for member in tf:
+ total_size += member.size
+ # Verify file by extraction
+ if member.isfile():
+ f = tf.extractfile(member)
+ if f is not None:
+ while True:
+ data = f.read(chunk_size)
+ if not data:
+ break
+ return total_size
+
+
+def verify_signature(pmc_name: str, artifact_path: str, signature_path: str) -> dict[str, Any]:
+ """Verify a signature file using the PMC's public signing keys."""
+ # Query only the signing keys associated with this PMC
+ with get_db_session() as session:
+ from sqlalchemy.sql.expression import ColumnElement
+
+ statement = (
+ select(PublicSigningKey)
+ .join(PMCKeyLink)
+ .join(PMC)
+ .where(cast(ColumnElement[bool], PMC.project_name == pmc_name))
+ )
+ result = session.execute(statement)
+ public_keys = [key.ascii_armored_key for key in result.scalars().all()]
+
+ with open(signature_path, "rb") as sig_file:
+ return verify_gpg_signature_file(sig_file, artifact_path, public_keys)
+
+
+def verify_gpg_signature_file(sig_file: BinaryIO, artifact_path: str, ascii_armored_keys: list[str]) -> dict[str, Any]:
+ """Verify a GPG signature for a file."""
+
+ @contextmanager
+ def ephemeral_gpg_home() -> Generator[str]:
+ """Create a temporary directory for an isolated GPG home, and clean it up on exit."""
+ temp_dir = tempfile.mkdtemp(prefix="gpg-")
+ try:
+ yield temp_dir
+ finally:
+ shutil.rmtree(temp_dir)
+
+ with ephemeral_gpg_home() as gpg_home:
+ gpg = gnupg.GPG(gnupghome=gpg_home)
+
+ # Import all PMC public signing keys
+ for key in ascii_armored_keys:
+ import_result = gpg.import_keys(key)
+ if not import_result.fingerprints:
+ # TODO: Log warning about invalid key?
+ continue
+ verified = gpg.verify_file(sig_file, str(artifact_path))
+
+ # Collect all available information for debugging
+ debug_info = {
+ "key_id": verified.key_id or "Not available",
+ "fingerprint": verified.fingerprint.lower() if verified.fingerprint else "Not available",
+ "pubkey_fingerprint": verified.pubkey_fingerprint.lower() if verified.pubkey_fingerprint else "Not available",
+ "creation_date": verified.creation_date or "Not available",
+ "timestamp": verified.timestamp or "Not available",
+ "username": verified.username or "Not available",
+ "status": verified.status or "Not available",
+ "valid": bool(verified),
+ "trust_level": verified.trust_level if hasattr(verified, "trust_level") else "Not available",
+ "trust_text": verified.trust_text if hasattr(verified, "trust_text") else "Not available",
+ "stderr": verified.stderr if hasattr(verified, "stderr") else "Not available",
+ "num_pmc_keys": len(ascii_armored_keys),
+ }
+
+ if not verified:
+ raise VerifyError("No valid signature found", debug_info)
+
+ return {
+ "verified": True,
+ "key_id": verified.key_id,
+ "timestamp": verified.timestamp,
+ "username": verified.username or "Unknown",
+ "email": verified.pubkey_fingerprint.lower() or "Unknown",
+ "status": "Valid signature",
+ "debug_info": debug_info,
+ }
+
+
def get_db_session() -> Session:
"""Get a new database session."""
+ # Create database engine
+ engine = create_engine("sqlite:///atr.db", echo=False)
return Session(engine)
@@ -92,22 +199,61 @@ def claim_next_task() -> tuple[int, str, str] | None:
return None
+def wrap(item: Any) -> tuple[Any, ...]:
+ """Ensure that returned results are structured as a tuple."""
+ if not isinstance(item, tuple):
+ return (item,)
+ return item
+
+
def process_task(task_id: int, task_type: str, task_args: str) -> None:
"""Process a claimed task."""
logger.info(f"Processing task {task_id} ({task_type}) with args {task_args}")
try:
- # TODO: Implement actual task processing
- time.sleep(1)
+ status = "COMPLETED"
+ if task_type == "verify_archive_integrity":
+ args = json.loads(task_args)
+ task_results = wrap(verify_archive_integrity(*args))
+ logger.info(f"Verified {args} and computed size {task_results[0]}")
+ elif task_type == "verify_signature":
+ args = json.loads(task_args)
+ task_results = wrap(verify_signature(*args))
+ logger.info(f"Verified {args} with result {task_results[0]}")
+ if task_results[0]["error"]:
+ status = "FAILED"
+ # Checksums are verified when uploaded, so we don't need to verify them here
+ # elif task_type == "verify_checksums":
+ # args = json.loads(task_args)
+ # task_results = wrap(verify_checksums(*args))
+ # logger.info(f"Verified {args} and computed size {task_results[0]}")
+ else:
+ error = f"Unknown task type: {task_type}"
+ logger.error(error)
+ raise Exception(error)
+ with get_db_session() as session:
+ result = json.dumps(task_results)
+ with session.begin():
+ session.execute(
+ text("""
+ UPDATE task
+ SET status = :status, completed = :now, result = :result
+ WHERE id = :task_id
+ """),
+ {"now": datetime.datetime.now(UTC), "task_id": task_id, "result": result, "status": status},
+ )
+ except VerifyError as e:
+ logger.error(f"Task {task_id} failed: {e.message}")
+ result = json.dumps(e.result)
with get_db_session() as session:
with session.begin():
session.execute(
text("""
UPDATE task
- SET completed = :now, status = 'COMPLETED'
+ SET status = 'FAILED', completed = :now, error = :error, result = :result
WHERE id = :task_id
"""),
- {"now": datetime.datetime.now(UTC), "task_id": task_id},
+ {"now": datetime.datetime.now(UTC), "task_id": task_id, "error": e.message, "result": result},
)
except Exception as e:
logger.error(f"Task {task_id} failed: {e}")
@@ -116,7 +262,7 @@ def process_task(task_id: int, task_type: str, task_args: str) -> None:
session.execute(
text("""
UPDATE task
- SET completed = :now, status = 'FAILED', error = :error
+ SET status = 'FAILED', completed = :now, error = :error
WHERE id = :task_id
"""),
{"now": datetime.datetime.now(UTC), "task_id": task_id, "error": str(e)},
@@ -125,12 +271,12 @@ def process_task(task_id: int, task_type: str, task_args: str) -> None:
def set_resource_limits() -> None:
"""Set CPU and memory limits for this process."""
- # Set CPU time limit
- try:
- resource.setrlimit(resource.RLIMIT_CPU, (CPU_LIMIT_SECONDS, CPU_LIMIT_SECONDS))
- logger.info(f"Set CPU time limit to {CPU_LIMIT_SECONDS} seconds")
- except ValueError as e:
- logger.warning(f"Could not set CPU time limit: {e}")
+ # # Set CPU time limit
+ # try:
+ # resource.setrlimit(resource.RLIMIT_CPU, (CPU_LIMIT_SECONDS, CPU_LIMIT_SECONDS))
+ # logger.info(f"Set CPU time limit to {CPU_LIMIT_SECONDS} seconds")
+ # except ValueError as e:
+ # logger.warning(f"Could not set CPU time limit: {e}")
# Set memory limit
try:
@@ -140,9 +286,10 @@ def set_resource_limits() -> None:
logger.warning(f"Could not set memory limit: {e}")
-def worker_loop() -> NoReturn:
+def worker_loop() -> None:
"""Main worker loop."""
- logger.info(f"Worker starting (PID: {os.getpid()})")
+ if os.path.isdir("state"):
+ os.chdir("state")
while True:
try:
@@ -150,6 +297,9 @@ def worker_loop() -> NoReturn:
if task:
task_id, task_type, task_args = task
process_task(task_id, task_type, task_args)
+ # Only process one task and then exit
+ # This prevents memory leaks from accumulating
+ break
else:
# No tasks available, wait 20ms before checking again
time.sleep(0.02)
@@ -177,4 +327,15 @@ def main() -> None:
if __name__ == "__main__":
- main()
+ with open("atr-worker-error.log", "a") as f:
+ f.write(f"{datetime.datetime.now(UTC)}: starting\n")
+ f.flush()
+ try:
+ main()
+ except Exception as e:
+ with open("atr-worker-error.log", "a") as f:
+ f.write(f"{datetime.datetime.now(UTC)}: {e}\n")
+ f.flush()
+ with open("atr-worker-error.log", "a") as f:
+ f.write(f"{datetime.datetime.now(UTC)}: exiting\n")
+ f.flush()
diff --git a/docs/plan.html b/docs/plan.html
index 0320150..f318786 100644
--- a/docs/plan.html
+++ b/docs/plan.html
@@ -11,6 +11,7 @@
<li>[DONE] Differentiate between podling PPMCs and top level PMCs</li>
<li>[DONE] Allow package deletion</li>
<li>[DONE] Allow RCs to be deleted</li>
+<li>[DONE] Move signature verification to a task runner</li>
</ul>
</li>
<li>
@@ -22,7 +23,8 @@
<li>[DONE] Show KB, MB, or GB units for file sizes</li>
<li>[DONE] Add a standard artifact naming pattern based on the project and product</li>
<li>[DONE] Potentially add the option to upload package artifacts without signatures</li>
-<li>Show validation status indicators</li>
+<li>[DONE] Show validation status indicators</li>
+<li>Improve validation status indicators</li>
<li>Add developer RC download buttons with clear verification instructions</li>
</ul>
</li>
@@ -31,8 +33,8 @@
<ul>
<li>[DONE] Display which PMCs are using each key</li>
<li>[DONE] Add key expiration warnings</li>
-<li>Fix reported problem with adding keys</li>
-<li>Add debugging output error messages for when key addition fails</li>
+<li>[DONE] Fix reported problem with adding keys</li>
+<li>[DONE] Add debugging output error messages for when key addition fails</li>
<li>Allow adding keys from a KEYS file</li>
</ul>
</li>
@@ -57,7 +59,7 @@
<li>
<p>Task runner workers</p>
<ul>
-<li>Implement worker process with RLIMIT controls for CPU and RAM</li>
+<li>[DONE] Implement worker process with RLIMIT controls for CPU and RAM</li>
<li>Add disk usage tracking through API and psutil polling</li>
<li>Add rollback or reporting for failed tasks</li>
<li>Ensure idempotent operations where possible</li>
@@ -68,12 +70,19 @@
<li>
<p>Orchestrating scheduler and resource management</p>
<ul>
-<li>Implement process-based task isolation</li>
-<li>Create task table in sqlite database</li>
-<li>Add task queue management</li>
+<li>[DONE] Implement process-based task isolation</li>
+<li>[DONE] Create task table in sqlite database</li>
+<li>[DONE] Add task queue management</li>
<li>Track and limit disk usage per task in the scheduler</li>
</ul>
</li>
+<li>
+<p>Improve the task UI</p>
+<ul>
+<li>Test that tasks are deleted when a package is deleted</li>
+<li>Allow restarting failed tasks</li>
+</ul>
+</li>
</ol>
<p>Advanced tasks, possibly deferred</p>
<ul>
@@ -83,11 +92,17 @@
<h2>Site improvements</h2>
<ol>
<li>
-<p>Ensure that performance is optimal and debugging is easy</p>
+<p>Fix bugs and improve workflow</p>
+<ul>
+<li>[DONE] Add ATR commit or version number to the UI</li>
+<li>Ensure that all errors are caught and logged or displayed</li>
+</ul>
+</li>
+<li>
+<p>Ensure that performance is optimal</p>
<ul>
<li>Add page load timing metrics to a log</li>
<li>Add a basic metrics dashboard</li>
-<li>Ensure that all errors are caught and logged or displayed</li>
</ul>
</li>
<li>
@@ -103,7 +118,8 @@
<li>
<p>Basic artifact validation</p>
<ul>
-<li>Implement basic structure validation (archives, signatures)</li>
+<li>[DONE] Implement basic archive verification</li>
+<li>[DONE] Implement basic signature verification</li>
</ul>
</li>
<li>
diff --git a/docs/plan.md b/docs/plan.md
index c0801fb..42ff8ef 100644
--- a/docs/plan.md
+++ b/docs/plan.md
@@ -11,6 +11,7 @@ This is a rough plan of immediate tasks. The priority of these tasks may change,
- [DONE] Differentiate between podling PPMCs and top level PMCs
- [DONE] Allow package deletion
- [DONE] Allow RCs to be deleted
+ - [DONE] Move signature verification to a task runner
2. Enhance RC display
- [DONE] Augment raw file hashes with the original filenames in the UI
@@ -19,7 +20,8 @@ This is a rough plan of immediate tasks. The priority of these tasks may change,
- [DONE] Show KB, MB, or GB units for file sizes
- [DONE] Add a standard artifact naming pattern based on the project and product
- [DONE] Potentially add the option to upload package artifacts without signatures
- - Show validation status indicators
+ - [DONE] Show validation status indicators
+ - Improve validation status indicators
- Add developer RC download buttons with clear verification instructions
3. Improve key management interface
@@ -44,7 +46,7 @@ Advanced tasks, possibly deferred
We aim to work on the task scheduler in parallel with the UX improvements above. Artifact validation and the release status dashboard are dependent on tasks, which are managed by the task scheduler.
1. Task runner workers
- - Implement worker process with RLIMIT controls for CPU and RAM
+ - [DONE] Implement worker process with RLIMIT controls for CPU and RAM
- Add disk usage tracking through API and psutil polling
- Add rollback or reporting for failed tasks
- Ensure idempotent operations where possible
@@ -52,11 +54,15 @@ We aim to work on the task scheduler in parallel with the UX improvements above.
- Test external tool use
2. Orchestrating scheduler and resource management
- - Implement process-based task isolation
- - Create task table in sqlite database
- - Add task queue management
+ - [DONE] Implement process-based task isolation
+ - [DONE] Create task table in sqlite database
+ - [DONE] Add task queue management
- Track and limit disk usage per task in the scheduler
+3. Improve the task UI
+ - Test that tasks are deleted when a package is deleted
+ - Allow restarting failed tasks
+
Advanced tasks, possibly deferred
- Check fair scheduling across cores
- Add task monitoring and reporting
@@ -64,7 +70,7 @@ Advanced tasks, possibly deferred
## Site improvements
1. Fix bugs and improve workflow
- - Add ATR commit or version number to the UI
+ - [DONE] Add ATR commit or version number to the UI
- Ensure that all errors are caught and logged or displayed
2. Ensure that performance is optimal
@@ -79,8 +85,8 @@ Advanced tasks, possibly deferred
These tasks are dependent on the task scheduler above.
1. Basic artifact validation
- - Implement basic archive verification
- - Implement basic signature verification
+ - [DONE] Implement basic archive verification
+ - [DONE] Implement basic signature verification
2. License compliance
- Verify LICENSE and NOTICE files exist and are placed correctly
diff --git a/migrations/versions/d3ff8c13bb8b_initial_schema.py b/migrations/versions/861b538fac8c_initial_schema.py
similarity index 84%
rename from migrations/versions/d3ff8c13bb8b_initial_schema.py
rename to migrations/versions/861b538fac8c_initial_schema.py
index a19128c..836006c 100644
--- a/migrations/versions/d3ff8c13bb8b_initial_schema.py
+++ b/migrations/versions/861b538fac8c_initial_schema.py
@@ -1,15 +1,15 @@
"""initial_schema
-Revision ID: d3ff8c13bb8b
+Revision ID: 861b538fac8c
Revises:
-Create Date: 2025-02-20 16:49:57.374714
+Create Date: 2025-02-21 16:38:15.050882
"""
from collections.abc import Sequence
# revision identifiers, used by Alembic.
-revision: str = "d3ff8c13bb8b"
+revision: str = "861b538fac8c"
down_revision: str | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org