258 lines
11 KiB
Python
258 lines
11 KiB
Python
"""Real PostgreSQL concurrency and transaction regressions for miner control."""
|
|
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime, timezone
|
|
from uuid import UUID
|
|
|
|
import psycopg
|
|
from psycopg.types.json import Jsonb
|
|
|
|
from app.agent_commands.models import AgentCommandResultRequest
|
|
from app.agent_commands.repository import AgentCommandRepository
|
|
from app.agent_commands.service import AgentCommandService
|
|
from app.config import settings
|
|
from app.miner_control.exceptions import MinerControlConflict
|
|
from app.miner_control.repository import MinerControlRepository
|
|
from app.miner_control.service import MinerControlService
|
|
|
|
|
|
SITE_ID = UUID("00000000-0000-4000-8000-000000001510")
|
|
ZONE_ID = UUID("00000000-0000-4000-8000-000000001514")
|
|
MINER_ID = UUID("00000000-0000-4000-8000-000000001511")
|
|
USER_ID = UUID("00000000-0000-4000-8000-000000001512")
|
|
OTHER_MINER_ID = UUID("00000000-0000-4000-8000-000000001513")
|
|
AGENT_ID = "control-test-agent"
|
|
|
|
|
|
def connect():
|
|
return psycopg.connect(
|
|
host=settings.postgres_host, port=settings.postgres_port,
|
|
dbname=settings.postgres_db, user=settings.postgres_user,
|
|
password=settings.postgres_password, connect_timeout=5,
|
|
)
|
|
|
|
|
|
def result_payload(accepted=True):
|
|
return {
|
|
"schema_version": 1,
|
|
"command_accepted": accepted,
|
|
"verification_attempted": False,
|
|
"verification_status": "unconfirmed" if accepted else "unavailable",
|
|
"observed_service_state": None,
|
|
"observed_performance_mode": None,
|
|
"failure_code": None if accepted else "write_rejected",
|
|
}
|
|
|
|
|
|
class MinerControlPostgresTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
connection = connect()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO sites (id, name) VALUES (%s, 'Control Test Site') "
|
|
"ON CONFLICT (id) DO NOTHING", (SITE_ID,)
|
|
)
|
|
cur.execute(
|
|
"INSERT INTO agent_zones (id, site_id, name) "
|
|
"VALUES (%s, %s, 'Control Test Zone') "
|
|
"ON CONFLICT (id) DO UPDATE SET is_enabled = true",
|
|
(ZONE_ID, SITE_ID),
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO users (id, username, display_name, password_hash, role_id)
|
|
SELECT %s, 'control-test-user', 'Control Test User', 'unused', id
|
|
FROM roles WHERE name = 'Admin' ON CONFLICT (id) DO NOTHING
|
|
""", (USER_ID,)
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miners (
|
|
id, site_id, agent_zone_id, asset_tag, current_ip, status, api3_supported,
|
|
configured_performance_mode, first_seen_at, last_seen_at
|
|
) VALUES (%s, %s, %s, 'CONTROL-TEST', '192.0.2.10', 'online', true,
|
|
'normal', NULL, NULL)
|
|
ON CONFLICT (id) DO UPDATE SET site_id = EXCLUDED.site_id,
|
|
agent_zone_id = EXCLUDED.agent_zone_id,
|
|
current_ip = EXCLUDED.current_ip, status = EXCLUDED.status,
|
|
api3_supported = true
|
|
""", (MINER_ID, SITE_ID, ZONE_ID)
|
|
)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO minerops_site_agents (
|
|
agent_id, site_id, agent_zone_id, site_name, last_payload
|
|
) VALUES (%s, %s, %s, 'Control Test Site', %s)
|
|
ON CONFLICT (agent_id) DO UPDATE SET site_id = EXCLUDED.site_id,
|
|
agent_zone_id = EXCLUDED.agent_zone_id,
|
|
last_payload = EXCLUDED.last_payload
|
|
""", (AGENT_ID, SITE_ID, ZONE_ID, Jsonb({"control_available": True}))
|
|
)
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
connection = connect()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute("DELETE FROM audit_logs WHERE entity_id = %s", (MINER_ID,))
|
|
cur.execute(
|
|
"DELETE FROM minerops_agent_command_queue "
|
|
"WHERE request_payload ->> 'miner_id' IN (%s, %s)",
|
|
(str(MINER_ID), str(OTHER_MINER_ID)),
|
|
)
|
|
cur.execute("DELETE FROM minerops_site_agents WHERE agent_id = %s", (AGENT_ID,))
|
|
cur.execute("DELETE FROM miners WHERE id = %s", (MINER_ID,))
|
|
cur.execute("DELETE FROM users WHERE id = %s", (USER_ID,))
|
|
cur.execute("DELETE FROM agent_zones WHERE id = %s", (ZONE_ID,))
|
|
cur.execute("DELETE FROM sites WHERE id = %s", (SITE_ID,))
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
def setUp(self):
|
|
connection = connect()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute("DELETE FROM audit_logs WHERE entity_id = %s", (MINER_ID,))
|
|
cur.execute(
|
|
"DELETE FROM minerops_agent_command_queue "
|
|
"WHERE request_payload ->> 'miner_id' IN (%s, %s)",
|
|
(str(MINER_ID), str(OTHER_MINER_ID)),
|
|
)
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
def insert_command(self, command_id, miner_id, status="pending", created=None):
|
|
connection = connect()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO minerops_agent_command_queue (
|
|
id, agent_id, command_type, command_status, request_payload, created_at
|
|
) VALUES (%s, %s, 'miner_control', %s, %s, COALESCE(%s, now()))
|
|
""",
|
|
(command_id, AGENT_ID, status,
|
|
Jsonb({"schema_version": 1, "miner_id": str(miner_id),
|
|
"action": "suspend", "requested_value": None}), created),
|
|
)
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
def test_two_concurrent_claimers_never_receive_same_command(self):
|
|
self.insert_command("concurrent-one", MINER_ID)
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = list(pool.map(
|
|
lambda _value: AgentCommandRepository(connect).claim_pending(AGENT_ID),
|
|
range(2),
|
|
))
|
|
claimed = [row["command_id"] for result in results for row in result]
|
|
self.assertEqual(claimed, ["concurrent-one"])
|
|
|
|
def test_skip_locked_claims_next_available_row(self):
|
|
self.insert_command(
|
|
"a-locked", MINER_ID,
|
|
created=datetime(2026, 8, 24, 10, tzinfo=timezone.utc),
|
|
)
|
|
self.insert_command(
|
|
"b-available", OTHER_MINER_ID,
|
|
created=datetime(2026, 8, 24, 11, tzinfo=timezone.utc),
|
|
)
|
|
locker = connect()
|
|
try:
|
|
with locker.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT id FROM minerops_agent_command_queue "
|
|
"WHERE id = 'a-locked' FOR UPDATE"
|
|
)
|
|
claimed = AgentCommandRepository(connect).claim_pending(AGENT_ID)
|
|
self.assertEqual([row["command_id"] for row in claimed], ["b-available"])
|
|
finally:
|
|
locker.rollback()
|
|
locker.close()
|
|
|
|
def test_result_binding_claim_requirement_and_terminal_immutability(self):
|
|
self.insert_command("result-one", MINER_ID, status="claimed")
|
|
request = AgentCommandResultRequest(
|
|
command_id="result-one", status="completed", result_payload=result_payload()
|
|
)
|
|
service = AgentCommandService(AgentCommandRepository(connect))
|
|
with self.assertRaises(Exception):
|
|
service.submit_result("wrong-agent", request)
|
|
service.submit_result(AGENT_ID, request)
|
|
with self.assertRaises(Exception):
|
|
service.submit_result(AGENT_ID, request)
|
|
|
|
self.insert_command("pending-result", OTHER_MINER_ID, status="pending")
|
|
pending = AgentCommandResultRequest(
|
|
command_id="pending-result", status="failed", result_payload=result_payload(False)
|
|
)
|
|
with self.assertRaises(Exception):
|
|
service.submit_result(AGENT_ID, pending)
|
|
|
|
def test_atomic_queue_audit_duplicate_live_and_completed_history(self):
|
|
service = MinerControlService(MinerControlRepository(connect))
|
|
first = service.queue_control(
|
|
MINER_ID, "suspend", requested_value=None,
|
|
actor_user_id=USER_ID, actor_username="control-test-user",
|
|
)
|
|
with self.assertRaises(MinerControlConflict):
|
|
service.queue_control(
|
|
MINER_ID, "suspend", requested_value=None,
|
|
actor_user_id=USER_ID, actor_username="control-test-user",
|
|
)
|
|
connection = connect()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT count(*) FROM audit_logs WHERE entity_id = %s", (MINER_ID,)
|
|
)
|
|
self.assertEqual(cur.fetchone()[0], 1)
|
|
cur.execute(
|
|
"UPDATE minerops_agent_command_queue SET command_status = 'completed', "
|
|
"completed_at = now() WHERE id = %s", (first.operation_id,)
|
|
)
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
second = service.queue_control(
|
|
MINER_ID, "suspend", requested_value=None,
|
|
actor_user_id=USER_ID, actor_username="control-test-user",
|
|
)
|
|
self.assertNotEqual(first.operation_id, second.operation_id)
|
|
|
|
def test_audit_failure_rolls_back_queue_row(self):
|
|
service = MinerControlService(MinerControlRepository(connect))
|
|
operation_count_before = self._operation_count()
|
|
with self.assertRaises(Exception):
|
|
service.queue_control(
|
|
MINER_ID, "suspend", requested_value=None,
|
|
actor_user_id=UUID("00000000-0000-4000-8000-000000009999"),
|
|
actor_username="missing-user",
|
|
)
|
|
self.assertEqual(self._operation_count(), operation_count_before)
|
|
|
|
def _operation_count(self):
|
|
connection = connect()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT count(*) FROM minerops_agent_command_queue "
|
|
"WHERE request_payload ->> 'miner_id' = %s", (str(MINER_ID),)
|
|
)
|
|
return cur.fetchone()[0]
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|