diff --git a/.changeset/tx-transaction-attempt-metadata.md b/.changeset/tx-transaction-attempt-metadata.md new file mode 100644 index 00000000000..8cb94990d76 --- /dev/null +++ b/.changeset/tx-transaction-attempt-metadata.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +`Effect.Transaction` now stores attempt metadata, analogous to `Schedule.CurrentMetadata`. Yield it inside a transaction body to read `attempt`, `retryReason`, and the same timing fields as a schedule (`start`, `now`, `elapsed`, `elapsedSincePrevious`). diff --git a/packages/effect/src/Effect.ts b/packages/effect/src/Effect.ts index b3139bfe933..b85e741ce7f 100644 --- a/packages/effect/src/Effect.ts +++ b/packages/effect/src/Effect.ts @@ -19,7 +19,7 @@ import type * as ExecutionPlan from "./ExecutionPlan.ts" import * as Exit from "./Exit.ts" import type { Fiber } from "./Fiber.ts" import type * as Filter from "./Filter.ts" -import { constant, dual, type LazyArg } from "./Function.ts" +import { dual, type LazyArg } from "./Function.ts" import type { TypeLambda } from "./HKT.ts" import type { Inspectable } from "./Inspectable.ts" import * as core from "./internal/core.ts" @@ -14471,48 +14471,107 @@ export const trackDuration: { // Transactions // ----------------------------------------------------------------------------- +/** + * Why the current transaction attempt started after a previous attempt did not commit. + * + * **Details** + * + * `"retry"` means the previous attempt called {@link txRetry}. `"conflict"` + * means an accessed `TxRef` changed before commit. The first attempt has no + * retry reason. + * + * @category models + * @since 4.0.0 + */ +export type TransactionRetryReason = "retry" | "conflict" + +/** + * Journal and attempt metadata for an Effect transaction. + * + * **When to use** + * + * Use to type the value provided by {@link Transaction}. + * + * **Details** + * + * `attempt` starts at `1` and counts each run of the outermost transaction body. + * `retryReason` is `"retry"` after {@link txRetry} and `"conflict"` after an + * accessed `TxRef` changes before commit; it is unset on the first attempt. + * `start`, `now`, `elapsed`, and `elapsedSincePrevious` are millisecond + * timestamps and durations for the current attempt, matching {@link Schedule} + * metadata. + * + * @see {@link Transaction} for the context service that provides this metadata + * @see {@link TransactionRetryReason} for the retry-reason tags recorded on later attempts + * + * @category models + * @since 4.0.0 + */ +export interface TransactionMeta { + readonly journal: ReadonlyMap< + TxRef, + { + readonly version: number + readonly value: any + } + > + readonly attempt: number + readonly retryReason: TransactionRetryReason | undefined + readonly start: number + readonly now: number + readonly elapsed: number + readonly elapsedSincePrevious: number +} + +/** @internal */ +export interface TransactionMetaInner { + retry: boolean + journal: Map< + TxRef, + { + readonly version: number + value: any + } + > + attempt: number + retryReason: TransactionRetryReason | undefined + start: number + now: number + elapsed: number + elapsedSincePrevious: number +} + /** * Service that holds the current transaction state. * * **Details** * - * It includes a journal that stores non-committed changes to `TxRef` values and - * a retry flag that records whether the transaction should be retried. + * Nested {@link tx} calls reuse this same service instance. Yield it inside a + * transaction body to read {@link TransactionMeta} fields such as `attempt` + * and `retryReason`. * - * **Example** (Building transactions) + * **Example** (Reading transaction metadata) * * ```ts import.meta.vitest * import { Effect } from "effect" * - * // Transaction class for software transactional memory operations - * const txEffect = Effect.gen(function*() { + * const program = Effect.tx(Effect.gen(function*() { * const tx = yield* Effect.Transaction - * // Use transaction for coordinated state changes - * return "Transaction complete" - * }) + * return [tx.attempt, tx.retryReason, tx.elapsed, tx.elapsedSincePrevious] + * })) * - * const runnable = Effect.provideService(txEffect, Effect.Transaction, { - * retry: false, - * journal: new Map() - * }) - * Effect.runSync(runnable) // => "Transaction complete" + * Effect.runSync(program) // => [1, undefined, 0, 0] * ``` * + * @see {@link tx} for the outermost transaction boundary that creates this service + * @see {@link TransactionMeta} for the journal and attempt metadata + * * @category services * @since 4.0.0 */ export class Transaction extends Context.Service< Transaction, - { - retry: boolean - readonly journal: Map< - TxRef, - { - readonly version: number - value: any - } - > - } + TransactionMeta >()("effect/Effect/Transaction") {} /** @@ -14565,45 +14624,66 @@ export const tx = ( effect: Effect ): Effect> => withFiber((fiber) => { - let state = Context.getOrUndefined(fiber.context, Transaction) + const state = Context.getOrUndefined(fiber.context, Transaction) if (state) { return effect as Effect> } + const clock = fiber.getRef(internal.ClockRef) + const now = clock.currentTimeMillisUnsafe() // Create transaction state only at the outermost boundary - state = { journal: new Map(), retry: false } + const mutable: TransactionMetaInner = { + journal: new Map(), + retry: false, + attempt: 1, + retryReason: undefined, + start: now, + now, + elapsed: 0, + elapsedSincePrevious: 0 + } let result: Exit.Exit | undefined - return uninterruptibleMask((restore) => - flatMap( + return uninterruptibleMask((restore) => { + const run = restore(effect).pipe( + provideService(Transaction, mutable), + tapCause(() => { + if (!mutable.retry) return void_ + return restore(awaitPendingTransaction(mutable)) + }), + exit + ) + return flatMap( whileLoop({ while: () => !result, - body: constant( - restore(effect).pipe( - provideService(Transaction, state), - tapCause(() => { - if (!state.retry) return void_ - return restore(awaitPendingTransaction(state)) - }), - exit - ) - ), + body: () => { + stampTransaction(mutable, clock.currentTimeMillisUnsafe()) + return run + }, step(exit: Exit.Exit) { - if (state.retry || !isTransactionConsistent(state)) { - return clearTransaction(state) + if (mutable.retry || !isTransactionConsistent(mutable)) { + mutable.attempt++ + mutable.retryReason = mutable.retry ? "retry" : "conflict" + return clearTransaction(mutable) } if (Exit.isSuccess(exit)) { - commitTransaction(fiber, state) + commitTransaction(fiber, mutable) } else { - clearTransaction(state) + clearTransaction(mutable) } result = exit } }), () => result! ) - ) + }) }) -const isTransactionConsistent = (state: Transaction["Service"]) => { +const stampTransaction = (state: TransactionMetaInner, now: number) => { + state.elapsedSincePrevious = now - state.now + state.now = now + state.elapsed = now - state.start +} + +const isTransactionConsistent = (state: TransactionMeta) => { for (const [ref, { version }] of state.journal) { if (ref.version !== version) { return false @@ -14612,7 +14692,7 @@ const isTransactionConsistent = (state: Transaction["Service"]) => { return true } -const awaitPendingTransaction = (state: Transaction["Service"]) => +const awaitPendingTransaction = (state: TransactionMeta) => suspend(() => { const key = {} const refs = Array.from(state.journal.keys()) @@ -14633,7 +14713,7 @@ const awaitPendingTransaction = (state: Transaction["Service"]) => }) }) -function commitTransaction(fiber: Fiber, state: Transaction["Service"]) { +function commitTransaction(fiber: Fiber, state: TransactionMeta) { for (const [ref, { value }] of state.journal) { if (value !== ref.value) { ref.version = ref.version + 1 @@ -14646,7 +14726,7 @@ function commitTransaction(fiber: Fiber, state: Transaction["S } } -function clearTransaction(state: Transaction["Service"]) { +function clearTransaction(state: TransactionMetaInner) { state.retry = false state.journal.clear() } @@ -14690,7 +14770,8 @@ function clearTransaction(state: Transaction["Service"]) { export const txRetry: Effect = flatMap( Transaction, (state) => { - state.retry = true + const mutable = state as TransactionMetaInner + mutable.retry = true return interrupt } ) diff --git a/packages/effect/src/TxRef.ts b/packages/effect/src/TxRef.ts index 4d2a9798d7c..5b921294e6c 100644 --- a/packages/effect/src/TxRef.ts +++ b/packages/effect/src/TxRef.ts @@ -12,6 +12,7 @@ * @since 4.0.0 */ import * as Effect from "./Effect.ts" +import type { TransactionMetaInner } from "./Effect.ts" import { dual } from "./Function.ts" import { pipeArguments } from "./Pipeable.ts" import type { Pipeable } from "./Pipeable.ts" @@ -173,10 +174,11 @@ export const modify: { Effect.Transaction.pipe( Effect.flatMap((state) => Effect.sync(() => { - if (!state.journal.has(self)) { - state.journal.set(self, { version: self.version, value: self.value }) + const inner = state as TransactionMetaInner + if (!inner.journal.has(self)) { + inner.journal.set(self, { version: self.version, value: self.value }) } - const current = state.journal.get(self)! + const current = inner.journal.get(self)! const [returnValue, next] = f(current.value) current.value = next return returnValue diff --git a/packages/effect/test/Effect.test.ts b/packages/effect/test/Effect.test.ts index ad49fa1cebe..28903b01be0 100644 --- a/packages/effect/test/Effect.test.ts +++ b/packages/effect/test/Effect.test.ts @@ -3433,6 +3433,96 @@ describe("Effect", () => { assert.strictEqual(transactionValue, 20) })) }) + + describe("transaction metadata", () => { + it.effect("tracks nested metadata across retries and conflicts", () => + Effect.gen(function*() { + const ref = TxRef.makeUnsafe(0) + const started = yield* Deferred.make() + const readyForConflict = yield* Deferred.make() + const resume = yield* Deferred.make() + const snapshots = yield* Ref.make< + Array<{ + readonly attempt: number + readonly retryReason: Effect.TransactionRetryReason | undefined + readonly start: number + readonly now: number + readonly elapsed: number + readonly elapsedSincePrevious: number + readonly nestedSame: boolean + }> + >([]) + + const fiber = yield* Effect.forkChild(Effect.tx(Effect.gen(function*() { + const outer = yield* Effect.Transaction + const nestedSame = yield* Effect.tx(Effect.gen(function*() { + const inner = yield* Effect.Transaction + return inner === outer + })) + yield* Ref.update(snapshots, (current) => [ + ...current, + { + attempt: outer.attempt, + retryReason: outer.retryReason, + start: outer.start, + now: outer.now, + elapsed: outer.elapsed, + elapsedSincePrevious: outer.elapsedSincePrevious, + nestedSame + } + ]) + yield* TxRef.get(ref) + if (outer.attempt === 1) { + yield* Deferred.succeed(started, undefined) + return yield* Effect.txRetry + } + if (outer.attempt === 2) { + yield* Deferred.succeed(readyForConflict, undefined) + yield* Deferred.await(resume) + } + }))) + + yield* Deferred.await(started) + yield* TestClock.adjust("1 second") + yield* TxRef.set(ref, 1) + yield* Deferred.await(readyForConflict) + yield* TestClock.adjust("500 millis") + yield* TxRef.set(ref, 2) + yield* Deferred.succeed(resume, undefined) + yield* Fiber.join(fiber) + const recorded = yield* Ref.get(snapshots) + + assert.deepStrictEqual(recorded, [ + { + attempt: 1, + retryReason: undefined, + start: 0, + now: 0, + elapsed: 0, + elapsedSincePrevious: 0, + nestedSame: true + }, + { + attempt: 2, + retryReason: "retry", + start: 0, + now: 1000, + elapsed: 1000, + elapsedSincePrevious: 1000, + nestedSame: true + }, + { + attempt: 3, + retryReason: "conflict", + start: 0, + now: 1500, + elapsed: 1500, + elapsedSincePrevious: 500, + nestedSame: true + } + ]) + })) + }) }) describe("Effect.fn", () => { diff --git a/packages/effect/test/Effectable.test.ts b/packages/effect/test/Effectable.test.ts index bd29da0ad3d..86b81d46fdc 100644 --- a/packages/effect/test/Effectable.test.ts +++ b/packages/effect/test/Effectable.test.ts @@ -29,7 +29,7 @@ describe("Effectable", () => { return Effect.fail(this.value) } } - + assert.deepStrictEqual(yield* Effect.exit(new FailingBox(1)), Exit.fail(1)) }))