Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 76 additions & 72 deletions api/computations/store.py
Original file line number Diff line number Diff line change
Expand Up @@ -241,13 +241,10 @@ def create_run(
with conn.cursor(cursor_factory=RealDictCursor) as cur:
_ensure_project_write_access(cur, project_id=project_id, user=user)
_enforce_submission_budget(cur, user_id=user.id, project_id=project_id, cluster_name=cluster.name)
_ensure_tool_catalog(
cur,
user=user,
project_id=project_id,
tool_name=tool_name,
cluster=cluster,
)
# Keeps this cluster's installation in the catalog; it never
# attaches anything to the project.
_ensure_tool_catalog(cur, tool_name=tool_name, cluster=cluster)
_ensure_tool_attached(cur, project_id=project_id, tool_name=tool_name)
if upstream_run_id:
cur.execute(
"""
Expand Down Expand Up @@ -734,37 +731,67 @@ def _insert_event(
)


def catalog_type_id(tool_name: str) -> str:
"""Experiment template id of a registered tool: computation_<tool>_<revision>."""
revision = TOOL_DEFINITIONS[tool_name].revision.replace("-", "_")
return f"computation_{tool_name}_{revision}"


def _ensure_tool_attached(cur, *, project_id: str, tool_name: str) -> None:
"""A project runs only the computational experiments it has attached.
Projects attach unit experiments, physical or computational, from the
library; submitting a run never attaches one implicitly.
"""
cur.execute("SELECT to_regclass('public.project_experiments') AS project_table")
if not cur.fetchone()["project_table"]:
return # minimal test databases have no project workflow table
cur.execute(
"""
SELECT 1 FROM project_experiments
WHERE project_id = %s AND type_id = %s AND status = 'active'
""",
(project_id, catalog_type_id(tool_name)),
)
if not cur.fetchone():
title = TOOL_DEFINITIONS[tool_name].title
raise ValueError(
f"Add the {title} unit experiment to this project before submitting it"
)


def _ensure_tool_catalog(
cur,
*,
user: PlatformUser,
project_id: str,
tool_name: str,
cluster: ClusterConfig,
) -> None:
"""Register computational equipment/type when the full DT schema exists.
"""Register a tool's installation (digital equipment) and template.
Catalog entries belong to the platform, not to a user: no owner and no
owning organization, so every signed-in user can find and attach them.
The computation tables can also be tested against a minimal database, so
this integration is deliberately conditional on the platform catalog
tables being present.
this is deliberately conditional on the catalog tables being present.
"""

cur.execute(
"SELECT to_regclass('public.equipment_metadata') AS equipment_table, "
"to_regclass('public.experiment_types') AS type_table, "
"to_regclass('public.project_experiments') AS project_table"
"to_regclass('public.experiment_types') AS type_table"
)
tables = cur.fetchone()
# One lock for every catalog write (startup registration and
# submissions), so their row locks are never taken in opposite orders.
# Submissions take the budget lock (783412, 1) first, then this one.
cur.execute("SELECT pg_advisory_xact_lock(783412, 2)")
equipment_table = tables["equipment_table"]
type_table = tables["type_table"]
project_table = tables["project_table"]
if not equipment_table or not type_table:
return
definition = TOOL_DEFINITIONS[tool_name]
# A versioned executable installation is digital equipment. The same
# scientific experiment can be compatible with several installations.
equipment_id = cluster.installation_id
type_id = f"computation_{tool_name}_{definition.revision.replace('-', '_')}"
type_id = catalog_type_id(tool_name)
schema = definition.input_model.model_json_schema()
required = set(schema.get("required", []))
parameters = []
Expand Down Expand Up @@ -835,12 +862,15 @@ def _ensure_tool_catalog(
) VALUES (
%s, %s, 'Computational Software', 'computational', 'Quantum ESPRESSO', %s,
'Purdue RCAC', %s, 'rcac_slurm', 'dashboard', 'none',
%s, %s, 'approved', %s, '[]'::jsonb, %s,
CURRENT_TIMESTAMP, %s, CURRENT_TIMESTAMP
NULL, '', 'approved', %s, '[]'::jsonb, %s,
CURRENT_TIMESTAMP, NULL, CURRENT_TIMESTAMP
)
ON CONFLICT (domain_id) DO UPDATE SET
equipment_name = EXCLUDED.equipment_name,
asset_kind = EXCLUDED.asset_kind,
-- Platform catalog: no user owns it, every organization sees it.
owner_id = NULL,
owner_org = '',
model = EXCLUDED.model,
description = EXCLUDED.description,
parameters_json = EXCLUDED.parameters_json,
Expand All @@ -852,11 +882,8 @@ def _ensure_tool_catalog(
f"Quantum ESPRESSO {cluster.qe_version} on {cluster.display_name}",
cluster.qe_version,
f"Versioned Quantum ESPRESSO executable on Purdue RCAC {cluster.display_name}.",
user.id,
user.organization,
Json(equipment_parameters),
Json(equipment_config),
user.id,
),
)
cur.execute(
Expand All @@ -867,13 +894,20 @@ def _ensure_tool_catalog(
visibility, version, author_name, author_email, owner_id, owner_org
) VALUES (
%s, %s, %s, %s, %s, %s, %s, '[]'::jsonb,
'shared', 1, %s, %s, %s, %s
'shared', 1, 'Digital Twin platform', '', NULL, ''
)
ON CONFLICT (id) DO UPDATE SET
name = EXCLUDED.name,
description = EXCLUDED.description,
type_parameters_json = EXCLUDED.type_parameters_json,
outputs_json = EXCLUDED.outputs_json,
-- Rows written by an earlier version belonged to the first
-- submitter; the catalog belongs to the platform.
visibility = 'shared',
author_name = EXCLUDED.author_name,
author_email = EXCLUDED.author_email,
owner_id = NULL,
owner_org = '',
updated_at = CURRENT_TIMESTAMP
""",
(
Expand All @@ -884,10 +918,6 @@ def _ensure_tool_catalog(
equipment_id,
Json(parameters),
Json(outputs),
user.name,
user.email,
user.id,
user.organization,
),
)
cur.execute(
Expand All @@ -902,49 +932,23 @@ def _ensure_tool_catalog(
""",
(type_id, equipment_id, cluster.name),
)
if not project_table:
return
upstream_definition = TOOL_DEFINITIONS["cellrelaxdft"]
upstream_type = (
"computation_cellrelaxdft_"
+ upstream_definition.revision.replace("-", "_")
if tool_name == "elasticdft"
else ""
)
cur.execute(
"""
INSERT INTO project_experiments (
project_id, type_id, added_by, status, sequence_index,
upstream_type_id, handoff_type, handoff_description,
output_mapping_json
)
SELECT %s, %s, %s, 'active',
COALESCE(MAX(sequence_index), 0) + 1, %s,
'automated_computational_job',
'Registered computation on Purdue RCAC through Globus Compute and Slurm.', %s
FROM project_experiments WHERE project_id = %s
ON CONFLICT (project_id, type_id) DO UPDATE SET
status = 'active',
upstream_type_id = EXCLUDED.upstream_type_id,
handoff_type = EXCLUDED.handoff_type,
handoff_description = EXCLUDED.handoff_description,
output_mapping_json = EXCLUDED.output_mapping_json,
updated_at = CURRENT_TIMESTAMP
""",
(
project_id,
type_id,
user.id,
upstream_type,
Json(
{
"execution_backend": "globus_compute_service_account",
"upstream_output": "structure_dict" if upstream_type else "",
"downstream_input": "upstream_execution_id"
if upstream_type
else "",
}
),
project_id,
),
)


def ensure_catalog() -> None:
"""Register every tool on every configured cluster (idempotent).
Runs when the API and the worker start, so projects can attach the
computational experiments before anyone has submitted a run.
"""
from computations.rcac import configured_clusters

conn = get_pg_worker_connection()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
for cluster in configured_clusters().values():
for tool_name in TOOL_DEFINITIONS:
_ensure_tool_catalog(cur, tool_name=tool_name, cluster=cluster)
conn.commit()
finally:
conn.close()

10 changes: 10 additions & 0 deletions api/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,16 @@ def _start_background_jobs() -> None:
start_proposal_job_sweeper()
start_notification_sweeper()

from computations.store import ensure_catalog, feature_enabled as computations_enabled

if computations_enabled():
# Register the computational experiments so projects can attach them
# before any run is submitted.
try:
ensure_catalog()
except Exception:
logger.warning("Could not register the computation catalog at startup", exc_info=True)


# ── Startup / shutdown ───────────────────────────────────────────────────────
@asynccontextmanager
Expand Down
82 changes: 82 additions & 0 deletions api/tests/test_computations_postgres_e2e.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,15 +182,67 @@ def configured_database():
INSERT INTO project_members (project_id, nanohub_user_id, role)
VALUES ('E2E-PROJECT', 'e2e-user', 'pi')
ON CONFLICT (project_id, nanohub_user_id) DO NOTHING;
INSERT INTO projects (id, name, access_mode) VALUES ('E2E-PHYSICAL', 'Physical only', 'private')
ON CONFLICT (id) DO NOTHING;
INSERT INTO project_members (project_id, nanohub_user_id, role)
VALUES ('E2E-PHYSICAL', 'e2e-user', 'pi')
ON CONFLICT (project_id, nanohub_user_id) DO NOTHING;
"""
)
conn.commit()
finally:
conn.close()
ensure_tables()
# As in production: the catalog is registered at startup, and a project
# attaches the computational experiments it wants to run.
from computations.store import catalog_type_id, ensure_catalog
from computations.schemas import TOOL_DEFINITIONS

ensure_catalog()
conn = _connection()
try:
with conn, conn.cursor() as cur:
for tool_name in TOOL_DEFINITIONS:
cur.execute(
"""
INSERT INTO project_experiments (project_id, type_id, added_by)
VALUES ('E2E-PROJECT', %s, 'e2e-user')
ON CONFLICT (project_id, type_id) DO UPDATE SET status = 'active'
""",
(catalog_type_id(tool_name),),
)
finally:
conn.close()
yield


def test_project_runs_only_attached_computational_experiments(client):
payload = {"cluster": "fake", "inputs": {"mp_id": "mp-1265", "structure_dict": {}}}
# A project with no computational experiment attached cannot submit one,
# and submitting does not attach it.
refused = client.post("/api/computations/projects/E2E-PHYSICAL/runs/cellrelaxdft", json=payload)
assert refused.status_code == 400, refused.text
assert "Add the Cell Relax DFT unit experiment" in refused.json()["detail"]
conn = _connection()
try:
with conn.cursor() as cur:
cur.execute("SELECT count(*) FROM project_experiments WHERE project_id = 'E2E-PHYSICAL'")
assert cur.fetchone() == (0,)
cur.execute("SELECT count(*) FROM computation_runs WHERE project_id = 'E2E-PHYSICAL'")
assert cur.fetchone() == (0,)
# Catalog entries belong to the platform, not to a submitter.
cur.execute(
"SELECT owner_id, owner_org FROM equipment_metadata WHERE domain_id = 'digital_qe_7_1_fake'"
)
assert cur.fetchone() == (None, "")
cur.execute(
"SELECT owner_id, owner_org FROM experiment_types WHERE id LIKE 'computation_%%' ORDER BY id"
)
assert cur.fetchall() == [(None, "")] * 3
finally:
conn.close()


def test_pilot_budget_and_claim_renewal(client, monkeypatch):
from computations.store import claim_run, renew_run_claim
monkeypatch.setenv("RCAC_ALLOWED_PROJECT_IDS", "E2E-PROJECT")
Expand Down Expand Up @@ -585,3 +637,33 @@ def test_archive_enrichment_persists_only_diagnostics(client, monkeypatch):
(run_id, ENRICHMENT_EVENT))
assert cur.fetchone()[0] == 1
conn.close()


def test_catalog_registration_returns_old_rows_to_the_platform(configured_database):
# Earlier versions made the first submitter own the catalog rows.
from computations.store import ensure_catalog

conn = _connection()
try:
with conn, conn.cursor() as cur:
cur.execute(
"UPDATE experiment_types SET owner_id = 'e2e-user', owner_org = 'Purdue University', "
"visibility = 'private' WHERE id = 'computation_cellrelaxdft_r99_dt2'"
)
cur.execute(
"UPDATE equipment_metadata SET owner_id = 'e2e-user', owner_org = 'Purdue University' "
"WHERE domain_id = 'digital_qe_7_1_fake'"
)
ensure_catalog()
with conn.cursor() as cur:
cur.execute(
"SELECT owner_id, owner_org, visibility FROM experiment_types "
"WHERE id = 'computation_cellrelaxdft_r99_dt2'"
)
assert cur.fetchone() == (None, "", "shared")
cur.execute(
"SELECT owner_id, owner_org FROM equipment_metadata WHERE domain_id = 'digital_qe_7_1_fake'"
)
assert cur.fetchone() == (None, "")
finally:
conn.close()
16 changes: 12 additions & 4 deletions web/app/projects/[id]/page.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import { useRouter } from "next/navigation";
import { cn } from "@/lib/utils";
import { useAuth } from "@/lib/auth-context";
import { ProjectComputations } from "@/components/project-computations";
import { computationalTemplateInfo, computationRunHref } from "@/lib/computation-templates";
import { computationalTemplateInfo, computationRunHref, type ComputationToolName } from "@/lib/computation-templates";
import {
deleteProject,
createPublicationRequest,
Expand Down Expand Up @@ -442,6 +442,12 @@ export default function ProjectDetailPage({ params }: { params: Promise<{ id: st
const displayedProjectExperiments = [...projectExperiments, ...implicitProjectExperiments]
.sort((a, b) => (a.sequence_index || 9999) - (b.sequence_index || 9999));
const attachedExperimentIds = new Set(displayedProjectExperiments.map((template) => template.id));
// A project runs only the computational experiments it has attached.
const attachedComputationTools = Array.from(new Set(
projectExperiments
.map((template) => computationalTemplateInfo(template).tool)
.filter((tool): tool is ComputationToolName => tool !== null),
));
const availableExperimentTypes = experimentTypes.filter((template) => !attachedExperimentIds.has(template.id));

const handleAddProjectExperiment = async () => {
Expand Down Expand Up @@ -1121,9 +1127,11 @@ export default function ProjectDetailPage({ params }: { params: Promise<{ id: st
)}
</div>

<Suspense fallback={null}>
<ProjectComputations projectId={project.id} />
</Suspense>
{attachedComputationTools.length > 0 && (
<Suspense fallback={null}>
<ProjectComputations projectId={project.id} allowedTools={attachedComputationTools} />
</Suspense>
)}

{/* Unit Experiments / Process */}
<div className="rounded-xl border border-[hsl(var(--border))] bg-[hsl(var(--card))] p-6">
Expand Down
2 changes: 1 addition & 1 deletion web/components/computational-faq.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ export function ComputationalFaq() {
keywords: "researcher student member owner administrator role permissions viewer PI forbidden restricted approved pilot project",
answer: <>
<p>No. <strong>Researcher</strong> is the normal student platform role. To submit or cancel a calculation, you must be a <strong>PI</strong> or <strong>member</strong> of the project; platform Administrators can also submit. A viewer of an open project can see its runs but cannot submit. <strong>Administrator</strong> is the highest platform role, used for managing users, registrations, and other platform settings.</p>
<p>Any project where you are the PI or a member can submit calculations. You do not need to provide an RCAC password or SSH key in the run form.</p>
<p>A project can run a computational experiment once it is attached to the project, the same way physical unit experiments are. On the project page, add <strong>Cell Relax DFT</strong>, <strong>Cell Relax DFT 2D</strong> or <strong>Elastic DFT</strong> under <strong>Unit Experiments</strong>; the <strong>Registered computational experiments</strong> form then appears with those tools. The project’s PI and members can submit. You do not need to provide an RCAC password or SSH key in the run form.</p>
</>,
},
{
Expand Down
Loading