genichinandClaude Opus 4.8
a4e263c8e7
Implement REQ-005~009 (dcy-002 slice 1) as an onion stack on top of the
BAT-001 pure domain:
- application: IntentService (admission + single-writer transition +
intent_id idempotency via durable intent_log)
- adapters/persistence: PostgreSQLSlotRepository + PostgreSQLIntentStore +
trex-core-owned migrations (daily_slot per Appendix A, daily_symbol_entries,
slot_transition_log, intent_log) + in-memory intent store
- adapters/api: FastAPI /v1 (healthz, intents/buy|sell|{id}, slots, capacity)
- runtime: trex-core-daemon entrypoint with lifespan migrate+preallocate
Verification (against local PostgreSQL 18): 74 passed, ruff clean, domain
import sweep shows zero web/db deps (NFR-1), 20-thread concurrency admits
exactly the idle count (NFR-2), idempotent resubmit causes no double
transition (NFR-3), EXPECTED_SLOT_FIELDS<->information_schema drift test
(realizes dcy-001 OQ-2's deferred DB-side check), and a daemon E2E smoke.
Align all four baseline docs to the new daemon/API/persistence layers.
BAT-002 -> release-candidate; REQ-005~009 -> Implemented. Full async outbox,
broker execution, safety floor, and strategy daemons remain deferred to a
follow-up Discovery.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""Concurrency / single-writer test for IntentService (REQ-007 NFR-2/NFR-3).
|
|
|
|
Many threads submit buy intents competing for a limited number of idle slots;
|
|
exactly that many must be admitted and no slot may be double-used.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from datetime import date
|
|
|
|
from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
|
|
from trex_core.application.intents import BuyIntent, IntentService
|
|
from trex_core.domain.slot import SlotService, SlotState
|
|
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
|
|
|
|
DAY = date(2026, 6, 3)
|
|
|
|
|
|
def test_concurrent_buys_admit_exactly_idle_count():
|
|
repo = InMemorySlotRepository()
|
|
svc = SlotService(repo)
|
|
intents = IntentService(svc, repo, InMemoryIntentStore())
|
|
svc.preallocate_daily_slots(DAY, 5) # only 5 idle slots
|
|
|
|
results: list[str] = []
|
|
results_lock = threading.Lock()
|
|
|
|
def worker(i: int) -> None:
|
|
d = intents.submit_buy(BuyIntent(intent_id=f"buy-{i}", trading_date=DAY, symbol="005930", qty=1))
|
|
with results_lock:
|
|
results.append(d.status)
|
|
|
|
threads = [threading.Thread(target=worker, args=(i,)) for i in range(20)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
admitted = [s for s in results if s == "admitted"]
|
|
rejected = [s for s in results if s == "rejected"]
|
|
assert len(admitted) == 5 # exactly the idle slots
|
|
assert len(rejected) == 15
|
|
# every non-idle slot is pending_buy exactly once (no double-use)
|
|
pending = [s for s in repo.list_slots(DAY) if s.status is SlotState.PENDING_BUY]
|
|
idle = [s for s in repo.list_slots(DAY) if s.status is SlotState.IDLE]
|
|
assert len(pending) == 5 and len(idle) == 0
|
|
|
|
|
|
def test_duplicate_intent_id_admits_once_under_concurrency():
|
|
repo = InMemorySlotRepository()
|
|
svc = SlotService(repo)
|
|
intents = IntentService(svc, repo, InMemoryIntentStore())
|
|
svc.preallocate_daily_slots(DAY, 5)
|
|
|
|
statuses: list[str] = []
|
|
lock = threading.Lock()
|
|
|
|
def worker() -> None:
|
|
d = intents.submit_buy(BuyIntent(intent_id="same-id", trading_date=DAY, symbol="005930"))
|
|
with lock:
|
|
statuses.append(d.status)
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(10)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# same intent_id submitted 10x concurrently -> exactly ONE slot transitioned
|
|
# (the core idempotency guarantee). Racing duplicates observe the winner's
|
|
# decision, which may still be 'processing' before finalize; never 'rejected'.
|
|
pending = [s for s in repo.list_slots(DAY) if s.status is SlotState.PENDING_BUY]
|
|
assert len(pending) == 1
|
|
assert all(s in ("admitted", "processing") for s in statuses)
|