|
1 | | -"""Concurrent uploads of the same file must not create duplicate registry rows (#191). |
| 1 | +"""Concurrent uploads leave one registry row per file and no partial state (#191, "F1"). |
2 | 2 |
|
3 | | -The UNIQUE constraints on csv_files(file_hash) and csv_files(file_name) |
4 | | -settle the race at the database level. This test verifies that two |
5 | | -simultaneous uploads of the same content yield one success and one |
6 | | -graceful rejection — not a duplicate row or partial state. |
| 3 | +(Not #178, which is also labelled F1 and is a different defect.) |
| 4 | +
|
| 5 | +The registry invariant is enforced by the database: UNIQUE (file_hash) on csv_files, |
| 6 | +and, since M5, a per-dataset advisory lock that serialises uploads of the same |
| 7 | +filename. Neither was covered by a test that makes uploads genuinely overlap. This |
| 8 | +releases several uploads together with a barrier and repeats the race over many |
| 9 | +rounds, because which interleaving wins is up to the scheduler. Whatever the |
| 10 | +interleaving, the invariants below must hold. |
| 11 | +
|
| 12 | +Three cases, because the guard differs: |
| 13 | +
|
| 14 | +* identical file, same name: the dataset lock serialises them; the later ones see |
| 15 | + the content hash and are refused. |
| 16 | +* identical content under DIFFERENT names: different datasets, so different locks; |
| 17 | + only the file_hash constraint and the csv_<hash> table name stand in the way. This |
| 18 | + is the real race, and where a loser used to get a generic "database error". |
| 19 | +* same name, different content: not a rejection since M5 (decision 2): each is the |
| 20 | + next version. Concurrency must still give each a distinct version. |
| 21 | +
|
| 22 | +Mutation check (from #191): drop the file_hash constraint and the different-names |
| 23 | +test fails — |
| 24 | +
|
| 25 | + ALTER TABLE csv_uploads.csv_files DROP CONSTRAINT csv_files_file_hash_key; |
| 26 | +
|
| 27 | +Restore it afterwards with `ADD CONSTRAINT csv_files_file_hash_key UNIQUE (file_hash)` |
| 28 | +(delete any duplicate rows the mutated run left first). |
7 | 29 | """ |
8 | 30 |
|
9 | 31 | from __future__ import annotations |
10 | 32 |
|
| 33 | +import hashlib |
11 | 34 | import os |
| 35 | +import threading |
12 | 36 | import unittest |
13 | 37 | import uuid |
14 | | -from concurrent.futures import ThreadPoolExecutor, as_completed |
| 38 | +from concurrent.futures import ThreadPoolExecutor |
15 | 39 |
|
16 | 40 | import pytest |
17 | | -from fastapi.testclient import TestClient |
| 41 | +from psycopg2 import sql |
| 42 | + |
| 43 | +from api.db import Conn |
| 44 | +from api.services.dynamic_loader import upload_dynamic |
| 45 | +from api.tenancy import uploads_schema |
| 46 | + |
| 47 | +ROUNDS = 40 |
| 48 | +RACERS = 4 |
18 | 49 |
|
19 | | -from api.main import app |
20 | 50 |
|
21 | | -_HELP = ("Start PostgreSQL and ensure api/ can reach it (PGHOST/PGPORT/PGUSER/" |
22 | | - "PGPASSWORD/PGDATABASE). See scripts/start-api.ps1.") |
| 51 | +def _race(uploads: list[tuple[str, str]]) -> list[dict]: |
| 52 | + """Run each (file_name, content) upload at once; return the results in order.""" |
| 53 | + gate = threading.Barrier(len(uploads)) |
| 54 | + |
| 55 | + def go(item): |
| 56 | + gate.wait(timeout=30) |
| 57 | + return upload_dynamic(item[0], item[1], None, False) |
| 58 | + |
| 59 | + with ThreadPoolExecutor(max_workers=len(uploads)) as pool: |
| 60 | + return list(pool.map(go, uploads)) |
| 61 | + |
| 62 | + |
| 63 | +def _registry(names: list[str], hashes: list[str]) -> list[tuple]: |
| 64 | + with Conn() as conn, conn.cursor() as cur: |
| 65 | + cur.execute(sql.SQL( |
| 66 | + "SELECT file_name, file_hash, table_name FROM {}.csv_files " |
| 67 | + "WHERE file_name = ANY(%s) OR file_hash = ANY(%s)").format(sql.Identifier(uploads_schema())), |
| 68 | + (names, hashes)) |
| 69 | + return cur.fetchall() |
| 70 | + |
| 71 | + |
| 72 | +def _table_rows(table: str) -> int: |
| 73 | + with Conn() as conn, conn.cursor() as cur: |
| 74 | + cur.execute(sql.SQL("SELECT count(*) FROM {}.{}").format( |
| 75 | + sql.Identifier(uploads_schema()), sql.Identifier(table))) |
| 76 | + return cur.fetchone()[0] |
23 | 77 |
|
24 | 78 |
|
25 | 79 | @pytest.mark.integration |
| 80 | +@pytest.mark.usefixtures("live_api", "registry_cleanup") |
26 | 81 | class RegistryRace(unittest.TestCase): |
27 | | - @classmethod |
28 | | - def setUpClass(cls): |
29 | | - cls.ctx = TestClient(app) |
30 | | - try: |
31 | | - cls.client = cls.ctx.__enter__() |
32 | | - except Exception as exc: # noqa: BLE001 |
33 | | - raise AssertionError( |
34 | | - f"Could not start the API: {type(exc).__name__}: {exc}\n{_HELP}" |
35 | | - ) from None |
36 | | - |
37 | | - @classmethod |
38 | | - def tearDownClass(cls): |
39 | | - cls.ctx.__exit__(None, None, None) |
40 | | - |
41 | 82 | def setUp(self): |
42 | 83 | self.tag = f"{os.getpid()}_{uuid.uuid4().hex[:8]}" |
43 | 84 |
|
44 | | - def _upload(self, file_name: str, content: str) -> dict: |
45 | | - body = {"fileName": file_name, "content": content} |
46 | | - return self.client.post("/api/csv/upload", json=body).json() |
47 | | - |
48 | | - def test_concurrent_identical_uploads_one_wins(self): |
49 | | - """Two threads upload the same content simultaneously. |
50 | | -
|
51 | | - One should succeed; the other should fail gracefully (duplicate |
52 | | - hash detected). The registry must contain exactly one row. |
53 | | - """ |
54 | | - fname = f"race_same_{self.tag}.csv" |
55 | | - content = f"a,b\n1,{self.tag}\n" |
56 | | - |
57 | | - results: list[dict] = [] |
58 | | - with ThreadPoolExecutor(max_workers=2) as pool: |
59 | | - futures = [pool.submit(self._upload, fname, content) for _ in range(2)] |
60 | | - for f in as_completed(futures): |
61 | | - results.append(f.result()) |
62 | | - |
63 | | - statuses = [r.get("status") for r in results] |
64 | | - ok_count = statuses.count("ok") |
65 | | - self.assertGreaterEqual(ok_count, 1, f"At least one upload must succeed: {results}") |
66 | | - self.assertLessEqual(ok_count + statuses.count("error"), 2) |
67 | | - |
68 | | - files_resp = self.client.get("/api/csv/files").json() |
69 | | - matches = [f for f in files_resp if f.get("file_name") == fname] |
70 | | - self.assertEqual(len(matches), 1, f"Registry must have exactly one row: {matches}") |
71 | | - |
72 | | - def test_concurrent_same_name_different_content(self): |
73 | | - """Two threads upload different content under the same filename. |
74 | | -
|
75 | | - Both are kept (M5): the per-dataset lock gives them versions 1 and 2, never the |
76 | | - same version, and neither replaces the other. |
77 | | - """ |
78 | | - fname = f"race_name_{self.tag}.csv" |
79 | | - content_a = f"a,b\nA,{self.tag}\n" |
80 | | - content_b = f"x,y\nB,{self.tag}\n" |
81 | | - |
82 | | - results: list[dict] = [] |
83 | | - with ThreadPoolExecutor(max_workers=2) as pool: |
84 | | - fa = pool.submit(self._upload, fname, content_a) |
85 | | - fb = pool.submit(self._upload, fname, content_b) |
86 | | - for f in as_completed([fa, fb]): |
87 | | - results.append(f.result()) |
88 | | - |
89 | | - files_resp = self.client.get("/api/csv/files").json() |
90 | | - matches = [f for f in files_resp if f.get("file_name") == fname] |
91 | | - self.assertEqual([r["status"] for r in results], ["ok", "ok"], results) |
92 | | - self.assertEqual(sorted(m["table_name"][-3:] for m in matches), ["_v1", "_v2"], matches) |
| 85 | + def _assert_one_winner(self, results: list[dict], round_no: int) -> tuple[int, dict]: |
| 86 | + ok = [(i, r) for i, r in enumerate(results) if r["status"] == "ok"] |
| 87 | + self.assertEqual(len(ok), 1, f"round {round_no}: expected exactly one winner, " |
| 88 | + f"got {[r['status'] for r in results]}") |
| 89 | + return ok[0] |
| 90 | + |
| 91 | + def _assert_attributable(self, res: dict, winner_name: str, round_no: int): |
| 92 | + """The loser is told it lost, and to whom: not a bare error and not a 500.""" |
| 93 | + self.assertEqual(res["status"], "duplicate_file", f"round {round_no}: {res}") |
| 94 | + self.assertEqual(res["existingFileName"], winner_name, f"round {round_no}: {res}") |
| 95 | + self.assertEqual(res["reason"], "content", f"round {round_no}: {res}") |
| 96 | + |
| 97 | + def test_identical_file_uploaded_concurrently_registers_once(self): |
| 98 | + for n in range(ROUNDS): |
| 99 | + name = f"race_same_{self.tag}_{n}.csv" |
| 100 | + content = f"a,b\n1,{self.tag}_{n}\n2,{self.tag}_{n}\n" |
| 101 | + digest = hashlib.sha256(content.encode()).hexdigest() |
| 102 | + self.registry_cleanup.append(name) |
| 103 | + |
| 104 | + results = _race([(name, content)] * RACERS) |
| 105 | + |
| 106 | + _, winner = self._assert_one_winner(results, n) |
| 107 | + for r in results: |
| 108 | + if r["status"] != "ok": |
| 109 | + self._assert_attributable(r, name, n) |
| 110 | + self.assertEqual(len(_registry([name], [digest])), 1, f"round {n}: registry rows") |
| 111 | + self.assertEqual(_table_rows(winner["tableName"]), 2, f"round {n}: rows duplicated or lost") |
| 112 | + |
| 113 | + def test_identical_content_under_different_names_registers_once(self): |
| 114 | + """The case where file_hash, not file_name or the dataset lock, is the guard.""" |
| 115 | + for n in range(ROUNDS): |
| 116 | + content = f"a,b\n1,{self.tag}_{n}\n2,{self.tag}_{n}\n" |
| 117 | + digest = hashlib.sha256(content.encode()).hexdigest() |
| 118 | + names = [f"race_hash_{self.tag}_{n}_{i}.csv" for i in range(RACERS)] |
| 119 | + self.registry_cleanup.extend(names) |
| 120 | + |
| 121 | + results = _race([(nm, content) for nm in names]) |
| 122 | + |
| 123 | + idx, winner = self._assert_one_winner(results, n) |
| 124 | + for i, r in enumerate(results): |
| 125 | + if i != idx: |
| 126 | + self._assert_attributable(r, names[idx], n) |
| 127 | + rows = _registry(names, [digest]) |
| 128 | + self.assertEqual(len(rows), 1, f"round {n}: registry rows {rows}") |
| 129 | + self.assertEqual(rows[0][0], names[idx]) |
| 130 | + self.assertEqual(_table_rows(winner["tableName"]), 2, f"round {n}: rows duplicated or lost") |
| 131 | + |
| 132 | + def test_same_name_different_content_gets_distinct_versions(self): |
| 133 | + """Since M5 a repeated filename is the next version, never a rejection. Concurrent |
| 134 | + uploads must not pick the same version number or lose one.""" |
| 135 | + for n in range(ROUNDS): |
| 136 | + name = f"race_name_{self.tag}_{n}.csv" |
| 137 | + uploads = [(name, f"a,b\n{i},{self.tag}_{n}_{i}\n") for i in range(RACERS)] |
| 138 | + self.registry_cleanup.append(name) |
| 139 | + |
| 140 | + results = _race(uploads) |
| 141 | + |
| 142 | + self.assertEqual([r["status"] for r in results], ["ok"] * RACERS, |
| 143 | + f"round {n}: every version must be stored, got {results}") |
| 144 | + self.assertEqual(sorted(r["version"] for r in results), list(range(1, RACERS + 1)), |
| 145 | + f"round {n}: versions must be distinct and gapless") |
| 146 | + self.assertEqual(len({r["dataset"] for r in results}), 1) |
| 147 | + digests = [hashlib.sha256(c.encode()).hexdigest() for _, c in uploads] |
| 148 | + self.assertEqual(len(_registry([name], digests)), RACERS, f"round {n}: registry rows") |
0 commit comments