From 890118a7f922431ccd524d842014232dea129a18 Mon Sep 17 00:00:00 2001 From: liveforpresent Date: Wed, 28 Oct 2026 12:38:44 +0900 Subject: [PATCH 1/2] feat(normalization): fence admission and create during joint cutover --- .../RestClientTradingNormalizationClient.kt | 6 +- .../JdbcTradingNormalizationStore.kt | 25 +++- .../JdbcTradingNormalizationTransitionGate.kt | 59 ++++++++ .../V10__trading_normalization_transition.sql | 103 ++++++++++++++ .../PostgresTradingReviewStoreTest.kt | 128 ++++++++++++++++-- .../TradingNormalizationDispatcher.kt | 14 +- .../TradingNormalizationResultProcessor.kt | 10 +- .../TradingNormalizationService.kt | 2 + .../port/TradingNormalizationPorts.kt | 3 +- .../TradingNormalizationTransitionGate.kt | 9 ++ 10 files changed, 333 insertions(+), 26 deletions(-) create mode 100644 tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationTransitionGate.kt create mode 100644 tradingreview/adapter/persistence/src/main/resources/db/migration/V10__trading_normalization_transition.sql create mode 100644 tradingreview/port/src/main/kotlin/com/refinvest/core/tradingreview/port/TradingNormalizationTransitionGate.kt diff --git a/tradingreview/adapter/compute/src/main/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClient.kt b/tradingreview/adapter/compute/src/main/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClient.kt index d73bc2b..226ffd1 100644 --- a/tradingreview/adapter/compute/src/main/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClient.kt +++ b/tradingreview/adapter/compute/src/main/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClient.kt @@ -77,13 +77,13 @@ class RestClientTradingNormalizationClient private constructor( ?: "IDEMPOTENCY_PAYLOAD_CONFLICT") 401, 403 -> NormalizationSubmission.Rejected("INTERNAL_AUTHENTICATION_FAILED") 400, 422 -> NormalizationSubmission.Rejected("INTERNAL_REQUEST_CONTRACT_INVALID") - in 500..599 -> NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE") + in 500..599 -> NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE", outcomeUnknown = true) else -> NormalizationSubmission.Rejected("INTERNAL_REQUEST_CONTRACT_INVALID") } } } catch (_: Exception) { - // Never log the request or an exception: transport errors can contain a signed URI. - NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE") + // The server may have accepted the exact identity before the response was lost. + NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE", outcomeUnknown = true) } } diff --git a/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationStore.kt b/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationStore.kt index 877d612..345d5ba 100644 --- a/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationStore.kt +++ b/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationStore.kt @@ -65,19 +65,34 @@ class JdbcTradingNormalizationStore(private val jdbc: JdbcTemplate, private val AND status IN ('DISPATCH_PENDING','SUBMITTED','RUNNING','RESULT_READY','PUBLISHING','RETRY_WAIT') ORDER BY created_at DESC LIMIT 1""", { rs, _ -> run(rs) }, sessionId.value).firstOrNull() - override fun claimNext(now: Instant, lease: Duration, workerId: String): NormalizationDispatch? = - claim("PENDING", now, lease, workerId) + override fun claimNext(now: Instant, lease: Duration, workerId: String): NormalizationDispatch? { + // Defense at claim, in addition to the permit held across claim/send/receipt. + return org.springframework.transaction.support.TransactionTemplate( + org.springframework.jdbc.datasource.DataSourceTransactionManager(requireNotNull(jdbc.dataSource))).execute { + jdbc.execute("SELECT pg_advisory_xact_lock_shared(hashtext(current_schema()), 28028)") + if (jdbc.queryForObject("SELECT paused FROM trading_normalization_transition WHERE singleton", Boolean::class.java) != false) + null + else claim("PENDING", now, lease, workerId) + } + } + + override fun holdUnknownCreate(dispatchId: String, leaseOwner: String, now: Instant): Boolean = + jdbc.update("""UPDATE trading_normalization_dispatches SET lease_owner=NULL, + lease_expires_at=NULL,last_failure_code='CREATE_OUTCOME_UNKNOWN',updated_at=? + WHERE id=?::uuid AND lease_owner=?::uuid AND status='PENDING'""", + Timestamp.from(now), dispatchId, leaseOwner) == 1 override fun claimSubmitted(now: Instant, lease: Duration, workerId: String): NormalizationDispatch? = claim("SUBMITTED", now, lease, workerId) private fun claim(status: String, now: Instant, lease: Duration, workerId: String): NormalizationDispatch? = jdbc.query("""WITH due AS (SELECT id FROM trading_normalization_dispatches - WHERE status=? AND next_attempt_at<=? AND + WHERE status=? AND (status<>'PENDING' OR create_started_at IS NULL) AND next_attempt_at<=? AND (lease_expires_at IS NULL OR lease_expires_at<=?) ORDER BY next_attempt_at,id LIMIT 1 FOR UPDATE SKIP LOCKED) UPDATE trading_normalization_dispatches d SET lease_owner=?::uuid, - lease_expires_at=?,attempt_count=attempt_count+1,updated_at=? + lease_expires_at=?,attempt_count=attempt_count+1,updated_at=?, + create_started_at=CASE WHEN d.status='PENDING' THEN clock_timestamp() ELSE d.create_started_at END FROM due WHERE d.id=due.id RETURNING d.*""", { rs, _ -> dispatch(rs) }, status, Timestamp.from(now), Timestamp.from(now), workerId, Timestamp.from(now.plus(lease)), Timestamp.from(now)).firstOrNull() @@ -185,7 +200,7 @@ class JdbcTradingNormalizationStore(private val jdbc: JdbcTemplate, private val override fun scheduleRetry(dispatchId: String, leaseOwner: String, code: String, next: Instant, now: Instant): Boolean { val count = jdbc.update("""UPDATE trading_normalization_dispatches SET - lease_owner=NULL,lease_expires_at=NULL,next_attempt_at=?,last_failure_code=?,updated_at=? + lease_owner=NULL,lease_expires_at=NULL,create_started_at=NULL,next_attempt_at=?,last_failure_code=?,updated_at=? WHERE id=?::uuid AND lease_owner=?::uuid AND lease_expires_at>? AND status='PENDING'""", Timestamp.from(next), code, Timestamp.from(now), dispatchId, leaseOwner, Timestamp.from(now)) if (count == 1) jdbc.update("""UPDATE trading_normalization_runs SET status='RETRY_WAIT', diff --git a/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationTransitionGate.kt b/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationTransitionGate.kt new file mode 100644 index 0000000..9b9efb3 --- /dev/null +++ b/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationTransitionGate.kt @@ -0,0 +1,59 @@ +package com.refinvest.core.tradingreview.adapter.persistence + +import com.refinvest.core.tradingreview.port.TradingFailure +import com.refinvest.core.tradingreview.port.TradingNormalizationTransitionGate +import org.springframework.jdbc.core.ConnectionCallback +import org.springframework.jdbc.core.JdbcTemplate +import org.springframework.stereotype.Repository +import org.springframework.transaction.support.TransactionSynchronizationManager +import java.util.concurrent.Semaphore + +@Repository +class JdbcTradingNormalizationTransitionGate(private val jdbc: JdbcTemplate) : TradingNormalizationTransitionGate { + // Reserve at most one connection per instance for the send fence. The fleet fence is PostgreSQL. + private val sending = Semaphore(1) + + override fun requireAdmission() { + if (!publicationAllowed()) throw TradingFailure("TEMPORARY_DEPENDENCY_FAILURE", 503) + } + + override fun publicationAllowed(): Boolean { + check(TransactionSynchronizationManager.isActualTransactionActive()) + jdbc.execute("SELECT pg_advisory_xact_lock_shared(hashtext(current_schema()), 28028)") + return jdbc.queryForObject("SELECT paused FROM trading_normalization_transition WHERE singleton", + Boolean::class.java) == false + } + + override fun withCreatePermit(action: () -> T): T? { + check(!TransactionSynchronizationManager.isActualTransactionActive()) + if (!sending.tryAcquire()) return null + try { + return jdbc.execute(ConnectionCallback { connection -> + connection.createStatement().use { + it.execute("SELECT pg_advisory_lock_shared(hashtext(current_schema()), 28028)") + } + try { + val open = connection.createStatement().use { statement -> + statement.executeQuery("SELECT paused FROM trading_normalization_transition WHERE singleton").use { + check(it.next()) + !it.getBoolean(1) + } + } + if (open) action() else null + } finally { + // Session advisory locks must never leak back into the connection pool. + try { + connection.createStatement().use { statement -> + statement.executeQuery("SELECT pg_advisory_unlock_shared(hashtext(current_schema()), 28028)").use { + check(it.next() && it.getBoolean(1)) + } + } + } catch (error: Exception) { + connection.abort { command -> command.run() } + throw error + } + } + }) + } finally { sending.release() } + } +} diff --git a/tradingreview/adapter/persistence/src/main/resources/db/migration/V10__trading_normalization_transition.sql b/tradingreview/adapter/persistence/src/main/resources/db/migration/V10__trading_normalization_transition.sql new file mode 100644 index 0000000..3d50a95 --- /dev/null +++ b/tradingreview/adapter/persistence/src/main/resources/db/migration/V10__trading_normalization_transition.sql @@ -0,0 +1,103 @@ +-- Internal DBA control. There is deliberately no public management endpoint. +CREATE TABLE trading_normalization_transition ( + singleton BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (singleton), + paused BOOLEAN NOT NULL DEFAULT FALSE, + epoch BIGINT NOT NULL DEFAULT 0 CHECK (epoch >= 0), + verified_evidence VARCHAR(240), + updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() +); +INSERT INTO trading_normalization_transition(singleton) VALUES (TRUE); +CREATE TABLE trading_normalization_transition_audit ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + epoch BIGINT NOT NULL, + action VARCHAR(30) NOT NULL, + actor TEXT NOT NULL, + evidence_ref VARCHAR(240) NOT NULL, + dispatch_id UUID, + occurred_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() +); +ALTER TABLE trading_normalization_dispatches ADD COLUMN create_started_at TIMESTAMPTZ; +-- An old pending attempt may already have reached Compute. Require reconciliation. +UPDATE trading_normalization_dispatches SET create_started_at=updated_at + WHERE status='PENDING' AND attempt_count>0; + +CREATE FUNCTION trading_normalization_transition_control( + command TEXT, expected_epoch BIGINT, evidence TEXT) RETURNS BIGINT +LANGUAGE plpgsql AS $$ +DECLARE state trading_normalization_transition%ROWTYPE; +BEGIN + IF evidence IS NULL OR evidence !~ '^[A-Za-z0-9][A-Za-z0-9._:/-]{0,239}$' THEN + RAISE EXCEPTION 'Safe evidence reference required'; + END IF; + PERFORM pg_advisory_xact_lock(hashtext(current_schema()), 28028); + SELECT * INTO STRICT state FROM trading_normalization_transition WHERE singleton FOR UPDATE; + IF state.epoch<>expected_epoch THEN RAISE EXCEPTION 'Transition epoch conflict'; END IF; + IF command='PAUSE' THEN + UPDATE trading_normalization_transition SET paused=TRUE,epoch=epoch+1, + verified_evidence=NULL,updated_at=clock_timestamp() WHERE singleton; + ELSIF command='VERIFY' AND state.paused THEN + IF EXISTS (SELECT 1 FROM trading_normalization_dispatches + WHERE status='PENDING' AND create_started_at IS NOT NULL) THEN + RAISE EXCEPTION 'Unresolved create outcome'; + END IF; + UPDATE trading_normalization_transition SET verified_evidence=evidence, + updated_at=clock_timestamp() WHERE singleton; + ELSIF command='RESUME' AND state.paused AND state.verified_evidence=evidence THEN + IF EXISTS (SELECT 1 FROM trading_normalization_dispatches + WHERE status='PENDING' AND create_started_at IS NOT NULL) THEN + RAISE EXCEPTION 'Unresolved create outcome'; + END IF; + UPDATE trading_normalization_transition SET paused=FALSE,updated_at=clock_timestamp() WHERE singleton; + ELSE RAISE EXCEPTION 'Transition not verified or invalid command'; + END IF; + INSERT INTO trading_normalization_transition_audit(epoch,action,actor,evidence_ref) + SELECT epoch,command,session_user,evidence FROM trading_normalization_transition WHERE singleton; + RETURN (SELECT epoch FROM trading_normalization_transition WHERE singleton); +END; +$$; +REVOKE ALL ON FUNCTION trading_normalization_transition_control(TEXT,BIGINT,TEXT) FROM PUBLIC; + +-- Only after an operator has compared real Compute status and its idempotency record. +-- Null Job ID is allowed solely for proven non-acceptance, preserving the original key. +CREATE FUNCTION trading_normalization_reconcile_create( + dispatch UUID, job TEXT, expected_epoch BIGINT, evidence TEXT) RETURNS VOID +LANGUAGE plpgsql AS $$ +DECLARE item trading_normalization_dispatches%ROWTYPE; +DECLARE parent trading_normalization_runs%ROWTYPE; +BEGIN + IF evidence IS NULL OR evidence !~ '^[A-Za-z0-9][A-Za-z0-9._:/-]{0,239}$' + OR (job IS NOT NULL AND (length(job)<1 OR length(job)>160)) THEN + RAISE EXCEPTION 'Safe reconciliation evidence required'; + END IF; + PERFORM pg_advisory_xact_lock(hashtext(current_schema()), 28028); + IF NOT EXISTS (SELECT 1 FROM trading_normalization_transition + WHERE singleton AND paused AND epoch=expected_epoch) THEN + RAISE EXCEPTION 'Reconciliation requires paused epoch'; + END IF; + SELECT * INTO STRICT item FROM trading_normalization_dispatches WHERE id=dispatch FOR UPDATE; + SELECT * INTO STRICT parent FROM trading_normalization_runs WHERE id=item.run_id FOR UPDATE; + IF item.status<>'PENDING' OR item.create_started_at IS NULL OR + parent.status NOT IN ('DISPATCH_PENDING','RETRY_WAIT') OR + NOT EXISTS (SELECT 1 FROM trading_import_sessions WHERE id=parent.import_session_id + AND NOT deletion_pending AND status='VALIDATING' AND active_normalization_run_id=parent.id + AND deletion_generation=parent.deletion_generation) THEN + RAISE EXCEPTION 'Create identity no longer current'; + END IF; + UPDATE trading_normalization_dispatches SET create_started_at=NULL, + status=CASE WHEN job IS NULL THEN 'PENDING' ELSE 'SUBMITTED' END, + lease_owner=NULL,lease_expires_at=NULL,last_failure_code=NULL, + next_attempt_at=clock_timestamp(),updated_at=clock_timestamp() WHERE id=dispatch; + IF job IS NOT NULL THEN + UPDATE trading_normalization_runs SET status='SUBMITTED',compute_job_id=job, + started_at=COALESCE(started_at,clock_timestamp()),updated_at=clock_timestamp(),version=version+1 + WHERE id=parent.id; + UPDATE trading_normalization_attempts SET compute_job_id=job + WHERE run_id=parent.id AND attempt_token=parent.attempt_token; + END IF; + UPDATE trading_normalization_transition SET verified_evidence=NULL WHERE singleton; + INSERT INTO trading_normalization_transition_audit(epoch,action,actor,evidence_ref,dispatch_id) + VALUES (expected_epoch,'RECONCILE_CREATE',session_user,evidence,dispatch); +END; +$$; +REVOKE ALL ON FUNCTION trading_normalization_reconcile_create(UUID,TEXT,BIGINT,TEXT) FROM PUBLIC; +REVOKE ALL ON trading_normalization_transition,trading_normalization_transition_audit FROM PUBLIC; diff --git a/tradingreview/adapter/persistence/src/test/kotlin/com/refinvest/core/tradingreview/adapter/persistence/PostgresTradingReviewStoreTest.kt b/tradingreview/adapter/persistence/src/test/kotlin/com/refinvest/core/tradingreview/adapter/persistence/PostgresTradingReviewStoreTest.kt index 510223f..bac7e42 100644 --- a/tradingreview/adapter/persistence/src/test/kotlin/com/refinvest/core/tradingreview/adapter/persistence/PostgresTradingReviewStoreTest.kt +++ b/tradingreview/adapter/persistence/src/test/kotlin/com/refinvest/core/tradingreview/adapter/persistence/PostgresTradingReviewStoreTest.kt @@ -365,7 +365,7 @@ class PostgresTradingReviewStoreTest { val processor = com.refinvest.core.tradingreview.application.TradingNormalizationResultProcessor( runs, imports, review, results, client, transactions, Clock.fixed(now, ZoneOffset.UTC), mapper, - object : TradingRetryJitter { override fun factor() = 0.5 }) + object : TradingRetryJitter { override fun factor() = 0.5 }, JdbcTradingNormalizationTransitionGate(jdbc)) assertTrue(processor.processOne()) assertEquals(1, descriptorCalls) assertEquals(if (mismatchedPayload) 1 else 0, downloadCalls) @@ -390,7 +390,7 @@ class PostgresTradingReviewStoreTest { } val processor = com.refinvest.core.tradingreview.application.TradingNormalizationResultProcessor( runs, imports, review, interrupted, client, transactions, Clock.fixed(now, ZoneOffset.UTC), mapper, - object : TradingRetryJitter { override fun factor() = 0.5 }) + object : TradingRetryJitter { override fun factor() = 0.5 }, JdbcTradingNormalizationTransitionGate(jdbc)) assertTrue(processor.processOne()) assertEquals(TradingImportStatus.VALIDATING, imports.find(session.id, MemberId(1))?.status) assertEquals(0L, jdbc.queryForObject("SELECT count(*) FROM trading_ledger_revisions", Long::class.java)) @@ -398,7 +398,7 @@ class PostgresTradingReviewStoreTest { val recoveredProcessor = com.refinvest.core.tradingreview.application.TradingNormalizationResultProcessor( runs, imports, review, results, client, transactions, Clock.fixed(now.plusSeconds(1), ZoneOffset.UTC), mapper, - object : TradingRetryJitter { override fun factor() = 0.5 }) + object : TradingRetryJitter { override fun factor() = 0.5 }, JdbcTradingNormalizationTransitionGate(jdbc)) assertTrue(recoveredProcessor.processOne()) assertEquals(1, descriptorCalls) assertEquals(1, downloadCalls) @@ -464,7 +464,7 @@ class PostgresTradingReviewStoreTest { val ackRecovered = com.refinvest.core.tradingreview.application.TradingNormalizationResultProcessor( runs, imports, review, results, client, transactions, Clock.fixed(now.plusSeconds(47), ZoneOffset.UTC), mapper, - object : TradingRetryJitter { override fun factor() = 0.5 }) + object : TradingRetryJitter { override fun factor() = 0.5 }, JdbcTradingNormalizationTransitionGate(jdbc)) assertTrue(ackRecovered.acknowledgeOne()) assertEquals(2, acknowledgeCalls) assertEquals("ACKNOWLEDGED", jdbc.queryForObject( @@ -522,7 +522,7 @@ class PostgresTradingReviewStoreTest { val processor = com.refinvest.core.tradingreview.application.TradingNormalizationResultProcessor( runs, imports, review, JdbcTradingNormalizationResultStore(jdbc, mapper), client, transactions, Clock.fixed(now, ZoneOffset.UTC), mapper, - object : TradingRetryJitter { override fun factor() = 0.5 }) + object : TradingRetryJitter { override fun factor() = 0.5 }, JdbcTradingNormalizationTransitionGate(jdbc)) assertTrue(processor.processOne()) assertEquals(TradingImportStatus.PROCESSING_FAILED, imports.find(session.id, MemberId(1))?.status) assertEquals("TERMINAL_FAILURE", jdbc.queryForObject( @@ -775,7 +775,7 @@ class PostgresTradingReviewStoreTest { } @Test - fun normalizationDispatchPersistsSnapshotAndRecoversExpiredLease() = withDatabase { connection -> + fun normalizationDispatchPersistsSnapshotAndHoldsUncertainExpiredLease() = withDatabase { connection -> val jdbc = JdbcTemplate(SingleConnectionDataSource(connection, true)) members(jdbc) val review = JdbcTradingReviewStore(jdbc) @@ -832,8 +832,7 @@ class PostgresTradingReviewStoreTest { assertEquals(null, store.claimNext(now, java.time.Duration.ofSeconds(45), UUID.randomUUID().toString())) val recovered = store.claimNext(now.plusSeconds(46), java.time.Duration.ofSeconds(45), UUID.randomUUID().toString()) - assertEquals(dispatch.id, recovered?.id) - assertEquals(2, recovered?.attemptCount) + assertNull(recovered) // The previous worker may already have reached Compute. jdbc.update("""UPDATE trading_normalization_dispatches SET status='SUBMITTED', lease_owner=NULL,lease_expires_at=NULL WHERE id=?::uuid""", dispatch.id) val pollWorker = UUID.randomUUID().toString() @@ -884,7 +883,23 @@ class PostgresTradingReviewStoreTest { java.time.Duration.ofSeconds(10), "NONE") override fun instrumentTermsJson() = "[]" override fun resultDeliveryPolicy() = deliveryPolicy - }, mapper, Clock.fixed(now, ZoneOffset.UTC)) + }, mapper, Clock.fixed(now, ZoneOffset.UTC), JdbcTradingNormalizationTransitionGate(jdbc)) + jdbc.execute("SELECT trading_normalization_transition_control('PAUSE',0,'test-validation')") + for (retry in listOf(false, true)) { + val failure = assertFailsWith { + transactions.execute { + if (retry) service.retry(session.id, "same-public-key") + else service.validate(session.id, null, "same-public-key") + } + } + assertEquals("TEMPORARY_DEPENDENCY_FAILURE", failure.code) + assertEquals(503, failure.httpStatus) + } + assertEquals(0L, jdbc.queryForObject("SELECT count(*) FROM trading_idempotency", Long::class.java)) + assertEquals(0L, jdbc.queryForObject("SELECT count(*) FROM trading_normalization_dispatches", Long::class.java)) + assertEquals(session, imports.find(session.id, MemberId(1))) + jdbc.execute("SELECT trading_normalization_transition_control('VERIFY',1,'test-validation-http')") + jdbc.execute("SELECT trading_normalization_transition_control('RESUME',1,'test-validation-http')") assertEquals("MISSING_REQUIRED_ARTIFACT", assertFailsWith { transactions.execute { service.validate(session.id, null, "missing-position") } }.code) @@ -896,7 +911,7 @@ class PostgresTradingReviewStoreTest { "AUTO", "objects/${UUID.randomUUID()}", byteArrayOf(1), "test-v1", TradingArtifactRetentionStatus.ACTIVE, now, now.plusSeconds(3600), null, null, 0)) } - val started = transactions.execute { service.validate(session.id, null, "validate-first") }!! + val started = transactions.execute { service.validate(session.id, null, "same-public-key") }!! assertEquals(TradingImportStatus.VALIDATING, started.status) assertNotNull(started.activeNormalizationRunId) val replay = transactions.execute { service.validate(session.id, null, "validate-again") }!! @@ -942,7 +957,7 @@ class PostgresTradingReviewStoreTest { val dispatcher = com.refinvest.core.tradingreview.application.TradingNormalizationDispatcher( runs, imports, review, imports, issuer, compute, object : TradingRetryJitter { override fun factor() = 0.5 - }, transactions, workerClock, mapper) + }, transactions, workerClock, mapper, JdbcTradingNormalizationTransitionGate(jdbc)) assertTrue(dispatcher.dispatchOne()) assertEquals(TradingImportStatus.VALIDATING, imports.find(session.id, MemberId(1))?.status) workerTime.set(now.plusSeconds(1)) @@ -991,7 +1006,7 @@ class PostgresTradingReviewStoreTest { val exhausting = com.refinvest.core.tradingreview.application.TradingNormalizationDispatcher( runs, imports, review, imports, issuer, alwaysRetry, object : TradingRetryJitter { override fun factor() = 0.5 - }, transactions, workerClock, mapper) + }, transactions, workerClock, mapper, JdbcTradingNormalizationTransitionGate(jdbc)) for (seconds in listOf(1L, 2L, 4L)) { workerTime.set(now.plusSeconds(seconds)) assertTrue(exhausting.dispatchOne()) @@ -1554,7 +1569,7 @@ class PostgresTradingReviewStoreTest { sessionId: TradingImportSessionId, attemptToken: String, deletionGeneration: Long): ArtifactReadGrant = error("poll only") }, compute, object : TradingRetryJitter { override fun factor() = 0.5 }, - transactions, Clock.fixed(now, ZoneOffset.UTC), mapper) + transactions, Clock.fixed(now, ZoneOffset.UTC), mapper, JdbcTradingNormalizationTransitionGate(jdbc)) assertTrue(dispatcher.pollOne(), code) val terminal = assertNotNull(imports.find(session.id, MemberId(1))) val expected = if (index < 6) TradingImportStatus.REJECTED else TradingImportStatus.PROCESSING_FAILED @@ -1569,6 +1584,85 @@ class PostgresTradingReviewStoreTest { } } + @Test + fun transitionFenceWaitsForAdmissionCommitAndSurvivesNewInstances() = withDatabase { connection -> + val source = transitionSource(connection.schema) + val jdbc = JdbcTemplate(source) + val gate = JdbcTradingNormalizationTransitionGate(jdbc) + val tx = TransactionTemplate(DataSourceTransactionManager(source)) + val entered = java.util.concurrent.CountDownLatch(1) + val release = java.util.concurrent.CountDownLatch(1) + val pool = Executors.newFixedThreadPool(2) + try { + val admission = pool.submit { + tx.execute { + gate.requireAdmission() + entered.countDown() + check(release.await(10, TimeUnit.SECONDS)) + } + } + assertTrue(entered.await(10, TimeUnit.SECONDS)) + val pause = pool.submit { + jdbc.queryForObject("SELECT trading_normalization_transition_control('PAUSE',0,'test-admission')", Long::class.java)!! + } + assertFailsWith { pause.get(200, TimeUnit.MILLISECONDS) } + release.countDown() + admission.get(10, TimeUnit.SECONDS) + assertEquals(1L, pause.get(10, TimeUnit.SECONDS)) + repeat(2) { + val restarted = JdbcTradingNormalizationTransitionGate(JdbcTemplate(transitionSource(connection.schema))) + val failure = assertFailsWith { tx.execute { restarted.requireAdmission() } } + assertEquals("TEMPORARY_DEPENDENCY_FAILURE", failure.code) + assertEquals(503, failure.httpStatus) + assertNull(restarted.withCreatePermit { error("No HTTP send while paused") }) + } + assertFailsWith { + jdbc.execute("SELECT trading_normalization_transition_control('RESUME',1,'unverified')") + } + jdbc.execute("SELECT trading_normalization_transition_control('VERIFY',1,'verified-http-evidence')") + jdbc.execute("SELECT trading_normalization_transition_control('RESUME',1,'verified-http-evidence')") + tx.execute { gate.requireAdmission() } + assertEquals(3L, jdbc.queryForObject("SELECT count(*) FROM trading_normalization_transition_audit", Long::class.java)) + } finally { release.countDown(); pool.shutdownNow() } + } + + @Test + fun transitionFenceWaitsForClaimSendAndReceiptAcrossWorkers() = withDatabase { connection -> + val jdbc = JdbcTemplate(transitionSource(connection.schema)) + val gate = JdbcTradingNormalizationTransitionGate(jdbc) + val entered = java.util.concurrent.CountDownLatch(1) + val release = java.util.concurrent.CountDownLatch(1) + val pool = Executors.newFixedThreadPool(2) + try { + val send = pool.submit { + gate.withCreatePermit { + entered.countDown() + check(release.await(10, TimeUnit.SECONDS)) + true + } ?: false + } + assertTrue(entered.await(10, TimeUnit.SECONDS)) + val pause = pool.submit { + jdbc.queryForObject("SELECT trading_normalization_transition_control('PAUSE',0,'test-send')", Long::class.java)!! + } + assertFailsWith { pause.get(200, TimeUnit.MILLISECONDS) } + release.countDown() + assertTrue(send.get(10, TimeUnit.SECONDS)) + assertEquals(1L, pause.get(10, TimeUnit.SECONDS)) + val worker2 = JdbcTradingNormalizationTransitionGate(JdbcTemplate(transitionSource(connection.schema))) + assertNull(worker2.withCreatePermit { error("Worker 2 must be stopped") }) + assertNull(gate.withCreatePermit { error("Worker 1 must be stopped") }) + } finally { release.countDown(); pool.shutdownNow() } + } + + private fun transitionSource(schema: String): org.springframework.jdbc.datasource.DriverManagerDataSource = + org.springframework.jdbc.datasource.DriverManagerDataSource().apply { + setUrl(requireNotNull(System.getenv("TRADING_REVIEW_TEST_JDBC_URL"))) + setUsername(requireNotNull(System.getenv("TRADING_REVIEW_TEST_DB_USER"))) + setPassword(requireNotNull(System.getenv("TRADING_REVIEW_TEST_DB_PASSWORD"))) + setConnectionProperties(java.util.Properties().apply { setProperty("currentSchema", schema) }) + } + private fun members(jdbc: JdbcTemplate) { jdbc.update("INSERT INTO members(id, role, created_at) VALUES (1, 'USER', CURRENT_TIMESTAMP)") jdbc.update("INSERT INTO members(id, role, created_at) VALUES (2, 'USER', CURRENT_TIMESTAMP)") @@ -1634,6 +1728,14 @@ class PostgresTradingReviewStoreTest { splitSqlStatements(migration9).forEach(statement::execute) } } + if (applyPinMigration) { + connection.createStatement().use { statement -> + val migration10 = javaClass.getResourceAsStream( + "/db/migration/V10__trading_normalization_transition.sql")!! + .bufferedReader().use { it.readText() } + splitSqlStatements(migration10).forEach(statement::execute) + } + } test(connection) } finally { connection.schema = "public" diff --git a/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationDispatcher.kt b/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationDispatcher.kt index 85098a1..a5e571f 100644 --- a/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationDispatcher.kt +++ b/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationDispatcher.kt @@ -30,8 +30,11 @@ class TradingNormalizationDispatcher( private val transactions: TransactionTemplate, private val clock: Clock, private val mapper: ObjectMapper, + private val transition: TradingNormalizationTransitionGate, ) { - fun dispatchOne(): Boolean { + fun dispatchOne(): Boolean = transition.withCreatePermit { dispatchPermitted() } ?: false + + private fun dispatchPermitted(): Boolean { val now = clock.instant() val workerId = UUID.randomUUID().toString() val dispatch = transactions.execute { @@ -61,10 +64,10 @@ class TradingNormalizationDispatcher( when (error.code) { "SOURCE_ARTIFACT_EXPIRED", "ARTIFACT_IDENTITY_CONFLICT", "INTERNAL_REQUEST_CONTRACT_INVALID" -> NormalizationSubmission.Rejected(error.code) - else -> NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE") + else -> NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE", outcomeUnknown = true) } } catch (_: Exception) { - NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE") + NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE", outcomeUnknown = true) } transactions.execute { val locked = sessions.find(run.input.importSessionId, MemberId(run.input.ownerId), forUpdate = true) @@ -78,6 +81,11 @@ class TradingNormalizationDispatcher( runs.markSubmitted(dispatch.id, workerId, run.id, run.input.deletionGeneration, run.attemptToken, submission.jobId, at) is NormalizationSubmission.Retry -> { + if (submission.outcomeUnknown) { + runs.holdUnknownCreate(dispatch.id, workerId, at) + runs.audit(run.id, "CREATE_OUTCOME_UNKNOWN", submission.code, at) + return@execute + } val policy = policy(run) if (dispatch.attemptCount >= policy.maxAttempts || Duration.between(run.createdAt, at) >= policy.maxElapsedTime) diff --git a/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationResultProcessor.kt b/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationResultProcessor.kt index f91a60f..a994697 100644 --- a/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationResultProcessor.kt +++ b/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationResultProcessor.kt @@ -26,6 +26,7 @@ class TradingNormalizationResultProcessor( private val clock: Clock, private val mapper: ObjectMapper, private val jitter: TradingRetryJitter, + private val transition: TradingNormalizationTransitionGate, @Value("\${refinvest.trading-review.normalization.delivery.parser-max-depth:64}") parserMaxDepth: Int = 64, @Value("\${refinvest.trading-review.normalization.delivery.parser-max-string-length:1048576}") @@ -200,7 +201,14 @@ class TradingNormalizationResultProcessor( as Map val independentlyVerified = stagedVerifier.verify(persistedBase, run, dispatch.id) comparePayloadDescriptor(independentlyVerified, descriptor) + var transitionHeld = false val published = transactions.execute { + if (!transition.publicationAllowed()) { + transitionHeld = true + results.retryReady(dispatch.id, worker, "NORMALIZATION_TRANSITION_PAUSED", + clock.instant().plusSeconds(5), clock.instant()) + return@execute false + } val locked = sessions.find(run.input.importSessionId, MemberId(run.input.ownerId), true) if (locked == null || !isCurrent(run, locked)) return@execute false if (!results.hasReadyLease(dispatch.id, worker, clock.instant())) return@execute false @@ -212,7 +220,7 @@ class TradingNormalizationResultProcessor( mapper.writeValueAsString(independentlyVerified.qualityBase), independentlyVerified.ledgerContentHash, clock.instant()) } == true - if (!published && transactions.execute { + if (!transitionHeld && !published && transactions.execute { results.hasReadyLease(dispatch.id, worker, clock.instant()) } == true) obsolete(run, dispatch, worker, descriptor) } catch (_: LeaseLost) { return true } diff --git a/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationService.kt b/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationService.kt index 15dd18e..7f558ae 100644 --- a/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationService.kt +++ b/tradingreview/application/src/main/kotlin/com/refinvest/core/tradingreview/application/TradingNormalizationService.kt @@ -26,6 +26,7 @@ open class TradingNormalizationService( private val settings: TradingNormalizationSettings, private val mapper: ObjectMapper, private val clock: Clock, + private val transition: TradingNormalizationTransitionGate, ) : TradingNormalizationUseCase { private val strictMapper = JsonMapper.builder() .enable(StreamReadFeature.STRICT_DUPLICATE_DETECTION).build() @@ -41,6 +42,7 @@ open class TradingNormalizationService( retry: Boolean): TradingImportSession { if (key.length !in 1..128 || !Regex("[A-Za-z0-9._~-]+").matches(key)) throw TradingFailure("IDEMPOTENCY_KEY_INVALID", 400) + transition.requireAdmission() val owner = members.currentMemberId() val initial = sessions.find(id, owner) ?: throw TradingFailure( if (sessions.deleted(id, owner)) "RESOURCE_DELETED" else "RESOURCE_NOT_FOUND", diff --git a/tradingreview/port/src/main/kotlin/com/refinvest/core/tradingreview/port/TradingNormalizationPorts.kt b/tradingreview/port/src/main/kotlin/com/refinvest/core/tradingreview/port/TradingNormalizationPorts.kt index a3700f5..da65028 100644 --- a/tradingreview/port/src/main/kotlin/com/refinvest/core/tradingreview/port/TradingNormalizationPorts.kt +++ b/tradingreview/port/src/main/kotlin/com/refinvest/core/tradingreview/port/TradingNormalizationPorts.kt @@ -15,6 +15,7 @@ interface TradingNormalizationStore { fun insert(run: NormalizationRun, dispatch: NormalizationDispatch) fun findRun(id: String): NormalizationRun? fun findActive(sessionId: TradingImportSessionId): NormalizationRun? + fun holdUnknownCreate(dispatchId: String, leaseOwner: String, now: Instant): Boolean fun claimNext(now: Instant, lease: Duration, workerId: String): NormalizationDispatch? fun claimSubmitted(now: Instant, lease: Duration, workerId: String): NormalizationDispatch? fun schedulePoll(dispatchId: String, leaseOwner: String, next: Instant, now: Instant): Boolean @@ -56,7 +57,7 @@ interface ArtifactAccessGrantIssuer { sealed interface NormalizationSubmission { data class Accepted(val jobId: String) : NormalizationSubmission - data class Retry(val code: String, val retryAfter: Duration? = null) : NormalizationSubmission + data class Retry(val code: String, val retryAfter: Duration? = null, val outcomeUnknown: Boolean = false) : NormalizationSubmission data class Rejected(val code: String) : NormalizationSubmission } diff --git a/tradingreview/port/src/main/kotlin/com/refinvest/core/tradingreview/port/TradingNormalizationTransitionGate.kt b/tradingreview/port/src/main/kotlin/com/refinvest/core/tradingreview/port/TradingNormalizationTransitionGate.kt new file mode 100644 index 0000000..bdc8bdb --- /dev/null +++ b/tradingreview/port/src/main/kotlin/com/refinvest/core/tradingreview/port/TradingNormalizationTransitionGate.kt @@ -0,0 +1,9 @@ +package com.refinvest.core.tradingreview.port + +/** Shared database fence. Admission/publication checks retain a lock until their transaction ends. */ +interface TradingNormalizationTransitionGate { + fun requireAdmission() + fun publicationAllowed(): Boolean + /** The permit spans durable claim, HTTP send and durable receipt; null means paused/busy. */ + fun withCreatePermit(action: () -> T): T? +} From c956391a5e9e2f1af9a552f081b5281c1b4efcbe Mon Sep 17 00:00:00 2001 From: liveforpresent Date: Wed, 28 Oct 2026 13:10:01 +0900 Subject: [PATCH 2/2] fix(normalization): preserve uncertain creates and document cutover operations --- .../TradingReviewFlywayMigrationTest.kt | 2 +- operations/NORMALIZATION_CUTOVER.md | 155 ++++++++++++++++++ .../RestClientTradingNormalizationClient.kt | 11 +- ...estClientTradingNormalizationClientTest.kt | 13 ++ .../JdbcTradingNormalizationStore.kt | 3 +- .../PostgresTradingReviewStoreTest.kt | 4 + 6 files changed, 183 insertions(+), 5 deletions(-) create mode 100644 operations/NORMALIZATION_CUTOVER.md diff --git a/app/src/test/kotlin/com/refinvest/TradingReviewFlywayMigrationTest.kt b/app/src/test/kotlin/com/refinvest/TradingReviewFlywayMigrationTest.kt index 6439d82..0194270 100644 --- a/app/src/test/kotlin/com/refinvest/TradingReviewFlywayMigrationTest.kt +++ b/app/src/test/kotlin/com/refinvest/TradingReviewFlywayMigrationTest.kt @@ -24,7 +24,7 @@ class TradingReviewFlywayMigrationTest { val flyway = Flyway.configure().dataSource(url, user, password).schemas(schema) .baselineOnMigrate(true).baselineVersion("0") .locations("classpath:db/migration").load() - assertEquals(8, flyway.migrate().migrationsExecuted) + assertEquals(10, flyway.migrate().migrationsExecuted) connection.schema = schema connection.createStatement().use { statement -> statement.executeQuery("SELECT count(*) FROM information_schema.tables WHERE table_schema='$schema' AND table_name IN ('trading_accounts','trading_books','trading_idempotency','trading_deletion_requests','trading_import_sessions','trading_source_artifacts')") diff --git a/operations/NORMALIZATION_CUTOVER.md b/operations/NORMALIZATION_CUTOVER.md new file mode 100644 index 0000000..aaf1610 --- /dev/null +++ b/operations/NORMALIZATION_CUTOVER.md @@ -0,0 +1,155 @@ +# Normalization joint cutover + +이 문서는 ADR-063/065/066/070, Context `TRADING_VERSIONING.md`의 공동 전환, +`TR-ACC-IMPORT-ORCH-028`을 실행하는 Core 운영 절차다. 공개 관리 API는 추가하지 않는다. +결과 버전 선택은 기존 V1/V1이며 V2 선택 활성화는 별도 작업이다. + +## Control and authority + +상태는 Core PostgreSQL schema의 `trading_normalization_transition` 한 행에 저장한다. +모든 participating Core 인스턴스는 같은 DB/schema를 사용해야 한다. 메모리 flag나 환경 변수의 +변경·재기동으로 상태를 바꾸지 않는다. `refinvest.compute.enabled`는 계속 true로 둔다. + +V10 migration owner 또는 승인된 DBA 운영 role만 아래 함수를 실행한다. 함수는 +`SECURITY INVOKER`이고 PUBLIC 실행 권한을 제거했다. 배포 시 애플리케이션 role은 +control/audit 테이블 SELECT만, 운영 role은 control UPDATE, audit INSERT와 sequence USAGE, +두 함수 EXECUTE를 갖도록 DBA가 부여한다. 운영 role에 reconcile을 허용하면 대상 +Run/attempt/dispatch SELECT·UPDATE 권한도 필요하다. runtime role을 migration owner로 +사용하지 않는다. 환경별 role 이름과 권한 부여는 인프라 소유 설정이다. + +성공한 PAUSE/VERIFY/RESUME/RECONCILE_CREATE는 DB `session_user`, epoch, 안전한 evidence +reference, DB 시각, 선택적 dispatch UUID를 감사 테이블에 저장한다. 공유 계정 대신 개인 운영 +identity를 사용하고 승인 ticket/배포 인벤토리/HTTP 검증 결과를 접근 통제된 변경 기록에 연결한다. +audit 테이블을 runtime에서 수정·삭제할 권한은 주지 않는다. 거절된 SQL 명령은 rollback되므로 +DB 접근 감사에도 남긴다. evidence에는 credential, signed URI, CSV/거래값을 넣지 않는다. + +## Fence boundary + +schema에 대해 `pg_advisory_*_lock(hashtext(current_schema()),28028)`을 사용한다. +admission은 public idempotency claim 전에 shared transaction lock을 얻고 Run/Session/dispatch/ +idempotency commit까지 유지한다. PAUSE의 exclusive transaction lock은 먼저 들어온 admission의 +commit을 기다린다. PAUSE가 commit된 뒤 시작한 정상 validation/retry는 기존 공개 오류 +`503 TEMPORARY_DEPENDENCY_FAILURE`, `retryable=true`이며 public claim과 상태 변경은 없다. + +create는 shared session lock을 durable claim부터 HTTP 응답·receipt 저장까지 유지한다. +이미 permit을 가진 send는 PAUSE 명령과 겹칠 수 있고 PAUSE의 완료 전에 끝난다. 성공적인 +PAUSE 함수 호출 **transaction commit**이 중지 완료 경계다. 명령을 제출한 시각을 완료로 +해석하지 않는다. 잠금 대기 timeout/connection loss는 중지 성공이 아니며 별도 연결로 상태를 +확인한다. Core가 DB와 연결을 잃은 시점의 네트워크 요청은 여전히 불명확 create로 대조한다. + +한 인스턴스의 동시 send permit은 1개다. fence용 connection과 claim/receipt transaction용 +connection이 필요하므로 runtime pool은 최소 2개, 정상 HTTP/poll 부하를 고려해 그 이상으로 +설정한다(현재 기본 pool 10개). session lock 해제 실패 시 연결을 abort하여 pool에 잠금을 +돌려주지 않는다. claim 자체에도 DB pause 검사를 둔다. scheduler on/off만으로 보호하지 않는다. + +PAUSE 동안 status/terminal 조회, grant 복구, source failure 처리, 삭제·취소·stale discard와 +이미 commit된 publication의 ACK 재시도를 유지한다. terminal 성공은 RESULT_READY/staging으로 +보존하고 새 publication/PERSISTED outbox 생성을 재개까지 보류한다. 이는 ORCH-028의 +`중지 중 신규 PERSISTED 없음`을 적용한다. 삭제 generation 검사가 항상 우선한다. + +## Stop and inspect + +1. 변경 ticket에 Context/Core/Compute/Web/Integration의 정확한 commit, 배포 instance ID, + image digest, schema/DB identity, 실행자와 승인 범위를 기록한다. 아래 SQL의 search_path를 + 실제 Core schema로 고정하고 DB identity를 확인한다. psql 접속 credential은 기존 secret + provider로 전달하고 command line/history/log에 넣지 않는다. + + ```sql + SELECT current_database(),current_schema(),session_user; + SELECT * FROM trading_normalization_transition; + SELECT epoch,action,actor,evidence_ref,dispatch_id,occurred_at + FROM trading_normalization_transition_audit ORDER BY id DESC LIMIT 30; + ``` + +2. 현재 epoch를 사용해 아래 명령을 autocommit 또는 명시적인 transaction commit으로 실행한다. + 이미 paused여도 새 PAUSE는 epoch를 증가시키고 과거 검증을 무효화한다. stale epoch는 거절된다. + + ```sql + SELECT trading_normalization_transition_control('PAUSE',0,'CHANGE-1234'); + -- 별도 연결에서 paused=true와 반환된 epoch=1을 확인한다. + ``` + +3. **처음 이 기능을 설치할 때 또는 이전 Core가 혼재하면** 위 DB gate만으로 해당 이전 + instance를 중지했다고 주장하지 않는다. 배포 인벤토리에서 gate 미지원 process를 제거하고, + 제거 완료까지 기존 ingress/egress 운영 제어로 validation/retry와 Normalization create를 + 차단한다. 그 사이 새 identity로 우회 전송하지 않는다. 이전 process/worker 0, old outbound + connection/queue 0, 계약 버전 필드 없는 미전송 producer 0의 근거를 변경 기록에 첨부한다. + 지원되는 인스턴스만 다시 기동하고 같은 DB의 paused 상태를 읽는지 확인한다. + +4. 미전송·불명확 작업과 보존 pin을 조사한다. 두 pin이 없는 legacy 배포에서는 그 배포의 + schema/provenance를 먼저 조사하고 V9/Core·0017/Compute backfill을 순서대로 수행한다. + + ```sql + SELECT d.id,d.run_id,d.compute_key,d.status,d.attempt_count,d.create_started_at, + d.lease_expires_at,n.compute_job_id,n.result_schema_version, + n.result_hash_specification_version,n.attempt_token,n.input_fingerprint + FROM trading_normalization_dispatches d JOIN trading_normalization_runs n ON n.id=d.run_id + WHERE d.status IN ('PENDING','SUBMITTED','RESULT_READY') ORDER BY d.created_at; + SELECT count(*) FROM trading_normalization_dispatches + WHERE result_schema_version IS NULL OR result_hash_specification_version IS NULL; + ``` + +## Drain, backfill and ambiguous creates + +이미 접수된 V1 Job은 status/terminal을 조회해 drain한다. 보존된 결과 metadata 또는 당시 V1-only +producer 계약을 근거로 Core V9와 Compute 0017을 적용한다. Run/Revision/result hash와 발급된 +dispatch key는 보존한다. Job request snapshot과 idempotency fingerprint/record를 함께 검증한다. +근거 없는 행에 V1을 채우거나 새 요청의 필드 누락을 default로 해석하지 않는다. + +durable claim은 `create_started_at`을 HTTP 전에 저장한다. 이후 timeout/5xx/429(필수 Retry-After가 있는 명시적 +503 TRADING_REVIEW_CAPACITY_EXCEEDED만 미접수로 확인)·응답 유실·Core crash이면 PENDING identity를 보류한다. lease 만료만으로 +재전송하지 않는다. 이전 PENDING attempt도 V10에서 보수적으로 이 상태로 전환한다. + +1. 기존 dispatch key의 SHA-256으로 **Compute 소유 idempotency record를 read-only** 대조한다. + 해당 Job의 실제 내부 HTTP GET status에서 Job ID, logical dispatch ID, Core Run ID, + attempt token, generation, input hash, VersionSet, V1/V1 pin을 Core와 비교한다. 기존 key로 + create를 보내서 존재 여부를 탐색하지 않는다. unknown 결과로 새 key/Run/Job을 만들지 않는다. +2. 일치하는 Job을 확인했으면 paused epoch에서 다음을 실행한다. Job bind는 기존 Run/attempt를 + SUBMITTED로 연결하고 status polling을 계속한다. 새 identity를 만들지 않는다. + + ```sql + SELECT trading_normalization_reconcile_create( + '00000000-0000-0000-0000-000000000001'::uuid, + 'verified-existing-compute-job-id',1,'CHANGE-1234-reconciled'); + ``` + +3. Compute의 durable idempotency 부재와 미접수를 모두 확인했을 때만 job 인자를 NULL로 하여 + 같은 dispatch key의 전송을 재개 가능하게 한다. 이 동작도 PAUSE를 해제하지 않으며 검증 + attest를 무효화한다. 확인할 수 없으면 보류 유지 후 incident 복구한다. 삭제되었거나 current가 + 아닌 identity는 reconcile 함수가 거절한다. Job 내용 대조는 운영자의 권한 있는 read-only + 검증 책임이며 Core DB 함수가 외부 HTTP 사실을 대신 증명하지 않는다. + +## Verify and resume + +1. 원래 fleet을 paused로 유지한다. 전용 비운영 DB와 실제 HTTP에서 배포할 exact Core/Compute + commit 조합을 실행한다. Integration의 result-version-pin runner는 별도 canary Core/Compute + schema에서 정상 JWT·CSRF, 합성 CSV 업로드, public validation과 실제 dispatcher로 V1/V1 + create를 생성한다. accepted/status/terminal/payload와 같은 key replay, 독립 V1 hash 검증, + 단일 Revision/commit 후 ACK를 확인한다. 숨겨진 default나 unsigned/contract 우회 요청을 쓰지 않는다. +2. 실패하면 VERIFY/RESUME을 실행하지 않는다. paused 상태를 별도 connection으로 확인하고, + 실패 로그·DB identity를 보존한다. 수정된 조합을 격리 환경에서 다시 검증한다. 혼합 버전 fleet에 + 신규 create를 보내지 않는다. 이전 binary rollback도 paused DB를 유지한 채 동일 pin·migration + 호환성을 확인한 뒤 수행한다. DB migration을 되돌리거나 Revision/hash를 덮어쓰지 않는다. +3. 성공 증적(정확한 commit과 instance/egress inventory 포함)을 접근 통제된 immutable 기록에 + 저장하고 그 SHA-256 reference를 다음 두 명령에 동일하게 사용한다. 아래는 형식 예시다. + + ```sql + SELECT trading_normalization_transition_control('VERIFY',1,'sha256/verified-evidence-digest'); + SELECT trading_normalization_transition_control('RESUME',1,'sha256/verified-evidence-digest'); + SELECT paused,epoch,verified_evidence FROM trading_normalization_transition; + ``` + + 함수는 unresolved PENDING create가 있거나 epoch/reference가 다르면 재개를 거절한다. + VERIFY는 DBA의 검증 완료 attest이며 arbitrary 문자열 자체가 HTTP 성공을 증명하지 않는다. + 재개 후 중지 중 거절했던 동일 public key로 validation/retry하고 단일 Run/Job/Revision/ACK를 + 확인한다. 실패 시 새 PAUSE로 epoch를 올리고 다시 조사한다. 대기 Run을 성공으로 처리하거나 + 미전송 dispatch를 삭제하지 않는다. + +## Migration runner limitation + +Compute 0017은 기존 terminal Job이 있는 schema에서 전체 SQL을 atomic transaction으로 실행하면 +deferred Job→terminal FK pending event와 ALTER TABLE이 충돌하는 알려진 SQLSTATE 55006이 있다. +Core gate/V10은 이 migration 오류를 고치지 않는다. Integration의 ORCH-027은 기존 psql statement별 +runner의 범위이며 atomic runner 배포의 통과 증거가 아니다. 해당 배포는 Compute 소유 SQL/runner의 +별도 해결·검증을 완료할 때까지 중지를 유지한다. 빈 canary schema에서 atomic migration이 성공한 +것도 기존 terminal Job의 atomic backfill 성공으로 해석하지 않는다. diff --git a/tradingreview/adapter/compute/src/main/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClient.kt b/tradingreview/adapter/compute/src/main/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClient.kt index 226ffd1..1eb7493 100644 --- a/tradingreview/adapter/compute/src/main/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClient.kt +++ b/tradingreview/adapter/compute/src/main/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClient.kt @@ -70,8 +70,15 @@ class RestClientTradingNormalizationClient private constructor( else if (pinError != null) NormalizationSubmission.Rejected(pinError) else NormalizationSubmission.Accepted(jobId) } - 503, 429 -> NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE", - response.headers.getFirst("Retry-After")?.toLongOrNull()?.let(Duration::ofSeconds)) + 503, 429 -> { + val failure = response.bodyTo(String::class.java)?.let(mapper::readTree) + val retrySeconds = response.headers.getFirst("Retry-After")?.toLongOrNull() + val definitelyRejected = response.statusCode.value() == 503 && + failure?.path("code")?.asText() == "TRADING_REVIEW_CAPACITY_EXCEEDED" && + retrySeconds != null && retrySeconds > 0 + NormalizationSubmission.Retry("TEMPORARY_DEPENDENCY_FAILURE", + retrySeconds?.let(Duration::ofSeconds), outcomeUnknown = !definitelyRejected) + } 409 -> NormalizationSubmission.Rejected( response.bodyTo(String::class.java)?.let { safeCode(it) } ?: "IDEMPOTENCY_PAYLOAD_CONFLICT") diff --git a/tradingreview/adapter/compute/src/test/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClientTest.kt b/tradingreview/adapter/compute/src/test/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClientTest.kt index 244bd1a..5607cb2 100644 --- a/tradingreview/adapter/compute/src/test/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClientTest.kt +++ b/tradingreview/adapter/compute/src/test/kotlin/com/refinvest/core/tradingreview/adapter/compute/RestClientTradingNormalizationClientTest.kt @@ -107,6 +107,15 @@ class RestClientTradingNormalizationClientTest { "targetVersionSet" to identity.targetVersionSet.toWireValue(), "resultSchemaVersion" to "TRADING_NORMALIZATION_RESULT_V2", "resultHashSpecificationVersion" to "TRADING_NORMALIZATION_RESULT_HASH_V2")))) + for ((status, body) in listOf( + HttpStatus.SERVICE_UNAVAILABLE to """{"code":"TRADING_REVIEW_CAPACITY_EXCEEDED"}""", + HttpStatus.SERVICE_UNAVAILABLE to "{}", + HttpStatus.BAD_GATEWAY to "{}", + HttpStatus.TOO_MANY_REQUESTS to "{}")) { + server.expect(requestTo("http://compute.test/trading-review/normalization-jobs")) + .andRespond(withStatus(status).contentType(MediaType.APPLICATION_JSON) + .header("Retry-After", "1").body(body)) + } val client = RestClientTradingNormalizationClient.withClient("test-internal-key", mapper, builder.build()) val grants = input.artifacts.map { artifact -> ArtifactReadGrant( artifact.artifactId.value.toString(), artifact.role.name, artifact.sha256, @@ -116,6 +125,10 @@ class RestClientTradingNormalizationClientTest { client.submit(run, dispatch, grants)).code) assertEquals("RESULT_VERSION_PIN_MISMATCH", assertIs( client.status(run, dispatch)).code) + for (unknown in listOf(false, true, true, true)) { + assertEquals(unknown, assertIs( + client.submit(run, dispatch, grants)).outcomeUnknown) + } server.verify() } diff --git a/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationStore.kt b/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationStore.kt index 345d5ba..59dfc02 100644 --- a/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationStore.kt +++ b/tradingreview/adapter/persistence/src/main/kotlin/com/refinvest/core/tradingreview/adapter/persistence/JdbcTradingNormalizationStore.kt @@ -69,8 +69,7 @@ class JdbcTradingNormalizationStore(private val jdbc: JdbcTemplate, private val // Defense at claim, in addition to the permit held across claim/send/receipt. return org.springframework.transaction.support.TransactionTemplate( org.springframework.jdbc.datasource.DataSourceTransactionManager(requireNotNull(jdbc.dataSource))).execute { - jdbc.execute("SELECT pg_advisory_xact_lock_shared(hashtext(current_schema()), 28028)") - if (jdbc.queryForObject("SELECT paused FROM trading_normalization_transition WHERE singleton", Boolean::class.java) != false) + if (jdbc.queryForObject("SELECT paused FROM trading_normalization_transition WHERE singleton FOR SHARE", Boolean::class.java) != false) null else claim("PENDING", now, lease, workerId) } diff --git a/tradingreview/adapter/persistence/src/test/kotlin/com/refinvest/core/tradingreview/adapter/persistence/PostgresTradingReviewStoreTest.kt b/tradingreview/adapter/persistence/src/test/kotlin/com/refinvest/core/tradingreview/adapter/persistence/PostgresTradingReviewStoreTest.kt index bac7e42..6a8ca3f 100644 --- a/tradingreview/adapter/persistence/src/test/kotlin/com/refinvest/core/tradingreview/adapter/persistence/PostgresTradingReviewStoreTest.kt +++ b/tradingreview/adapter/persistence/src/test/kotlin/com/refinvest/core/tradingreview/adapter/persistence/PostgresTradingReviewStoreTest.kt @@ -1638,6 +1638,10 @@ class PostgresTradingReviewStoreTest { gate.withCreatePermit { entered.countDown() check(release.await(10, TimeUnit.SECONDS)) + // PAUSE is already queued. Claim uses another connection and must not + // request another advisory lock behind that exclusive waiter. + val store = JdbcTradingNormalizationStore(jdbc, tools.jackson.databind.json.JsonMapper.builder().build()) + assertNull(store.claimNext(now, java.time.Duration.ofSeconds(45), UUID.randomUUID().toString())) true } ?: false }