diff --git a/CHANGELOG.md b/CHANGELOG.md index e17f770..1dedd8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,9 @@ All notable changes are documented here. Format based on [Keep a Changelog](http - Direct unit tests for `resolvePassphrase()` covering explicit passphrase present/blank, named network known/unknown, and neither-provided branches (#462) - `StreamsModule.forceCancel()` — wraps the contract's `force_cancel()` so a recipient can force-cancel a stream paused beyond the 30-day threshold (previously only a prose TODO; `StreamErrorCode.PauseThresholdNotMet` is now reachable through the SDK) (#453) - `StreamsModule.transferRecipient()` — wraps the contract's `transfer_recipient()` so the current recipient can reassign the recipient role (previously only a prose TODO) (#454) +- `ConduitConfig.fee` / `ConduitConfig.feeMultiplier` — configure the inclusion (bid) fee for submitted transactions, via `resolveFee()` (exported from `src/soroban.ts`) and a new `fee` parameter on `buildContractCallTx()`. Previously every transaction hardcoded `fee: BASE_FEE` with no override, so under inclusion-fee pressure a 100-stroop bid would never be selected (#509) +- `StreamEventHandlers.onCreated`, `.onForceCancel`, `.onRecipientTransfer`, `.onOperatorSet`, `.onOperatorRevoke` — `dispatchEvent()` previously silently ignored the `created`, `force_cxl`, `xfer_rec`, `set_op`, and `rm_op` contract topics, so a live subscriber was never notified when the recipient role was transferred, a paused stream was force-cancelled by the recipient, or an operator was delegated/revoked (#506) +- `Module44` stream liquidity-risk / runway calculator for Feature #44 (#378); follows `Module26`/`Module36`/`Module48`'s shared `LruMemoCache` pattern and reports an honest, workload-dependent `measuredSpeedupPercent` rather than a fixed percentage. Classifies each stream's remaining runway into `'inactive' | 'critical' | 'warning' | 'healthy'` and exposes `estimateTopUpNeeded()` to compute the stroops needed to reach a target runway - `NonceManager` (the bigint-based, queue-with-cancellation implementation) is now exported from the package entry point, along with its `NonceLock`/`NonceManagerOptions` types. Previously neither `NonceManager` implementation was reachable outside SDK source (#483). - `subscribeToStream()` / `client.streams.subscribe()` now accept `maxBackoffMs` and `maxConsecutiveFailures` options controlling the new exponential-backoff-with-cutoff polling behaviour (#485). - `paramToScVal(value, type?)` accepts an explicit ScVal type hint (`'u64'`, `'i128'`, ...) and passes already-encoded `xdr.ScVal`s through untouched; `BatchOperation.types` supplies per-field type information for the `params` map (e.g. `{ streamId: 'u64', amount: 'i128' }`) so a u64 stream ID or an i128 amount encodes with the exact width the contract expects instead of the default inference (#497). @@ -67,6 +70,8 @@ All notable changes are documented here. Format based on [Keep a Changelog](http - `catchNetworkError()` no longer reclassifies *any* `TypeError` whose text happens to contain `fetch`/`connect`/`network`/etc. It now only reclassifies errors that are provably transport failures: the canonical fetch/axios network messages (`fetch failed`, `Failed to fetch`, `Network Error`, `Load failed`) or an error (or its nested `cause`) carrying a network errno code such as `ECONNREFUSED`/`ENOTFOUND`/`ERR_NETWORK`. A programming `TypeError` (e.g. `Cannot read properties of undefined (reading 'connect')`) is re-thrown as-is instead of being masked as a network outage (#457). - `NonceManager` now throws a descriptive error for an unparseable nonce string (e.g. `startNonce: 'not-a-number'`) instead of silently coercing it to `0n`, which masked caller bugs as an explicit zero (#458). - `StreamBuilder.build()` now stringifies a numeric `ratePerSecond` so the runtime value matches the declared `ratePerSecond?: string` return type; previously a `number` input passed through unchanged, so callers trusting the type (`.trim()`, string concatenation) hit runtime errors (#459). +- `StreamsModule.batchWithdraw()` now submits withdrawals one at a time instead of firing them all concurrently via `Promise.allSettled(...map)`. Concurrent submission meant every withdrawal read the same account sequence number from `getAccount()` and built a transaction with the same value — with a single keypair/wallet, at most one of the N withdrawals could ever land, the rest failing with `txBAD_SEQ`, defeating the whole purpose of a batch (#504) +- `src/streams.ts` was missing its `Signer` type import, so `npm run build` / `npm run typecheck` failed unconditionally on `main` (`Cannot find name 'Signer'`) — added the missing `import type { Signer } from './signer.js'`. - `StreamsModule.withdraw()` / `topUp()` now reject `amount <= 0n` client-side (before any RPC round-trip), matching `create()`'s fail-fast validation philosophy instead of relying on the contract's `InvalidAmount` simulate+reject cycle (#451) - `StreamsModule.list()` no longer silently drops `recipient` when both `sender` and `recipient` are provided — it now returns the de-duplicated union of both filters (#452) diff --git a/docs/api.md b/docs/api.md index 15b4f76..fe65ed3 100644 --- a/docs/api.md +++ b/docs/api.md @@ -20,6 +20,17 @@ new ConduitClient(config: ConduitConfig) | `factoryAddress` | `string` | | Deployed factory | | `governorAddress` | `string` | | Deployed governor | | `wallet` | `WalletAdapter` | | — | +| `fee` | `string` | | Explicit inclusion (bid) fee in stroops for submitted transactions. Takes precedence over `feeMultiplier`. Defaults to `BASE_FEE` (100 stroops) | +| `feeMultiplier` | `number` | | Multiplier applied to `BASE_FEE` to compute the inclusion fee, e.g. `10` bids 10x the network minimum. Ignored when `fee` is set | + +> **Inclusion fee.** Every submitted transaction previously used `BASE_FEE` +> (the network minimum, 100 stroops) unconditionally, with no way to raise +> it. Under inclusion-fee pressure (surge pricing, congested ledgers) a +> 100-stroop bid is not selected, and the SDK's confirmation polling +> eventually throws a misleading `Transaction timed out` instead of +> surfacing "fee too low". Set `fee` (an exact stroops amount) or +> `feeMultiplier` (a multiple of `BASE_FEE`) on `ConduitConfig` to bid +> higher; `fee` wins if both are set. ### Convenience methods @@ -86,6 +97,38 @@ const total = await client.streams.streamedTotal(streamId); --- +### `batchWithdraw(withdrawals) → Promise` + +| Param | Type | Notes | +|-------|------|-------| +| `withdrawals` | `{ streamId: bigint \| string; amount?: bigint }[]` | One entry per stream to withdraw from | + +```typescript +interface BatchWithdrawResult { + streamId: bigint; + success: boolean; + txHash?: string; // present when success is true + error?: string; // present when success is false +} +``` + +Withdraws from multiple streams. Soroban permits only one +`invoke_host_function` operation per transaction, so this cannot be a single +atomic transaction — each withdrawal is submitted as its own transaction and +reported independently, so a failure on one streamId (e.g. +`StreamNotFound`, insufficient balance) does not fail the others. + +Withdrawals are submitted **one at a time, not concurrently**. Each +transaction is built from the caller account's current sequence number; +submitting all N withdrawals at once means every one of them would read the +same sequence number and collide, so with a single keypair/wallet at most +one could ever land on-chain. Awaiting each withdrawal in turn guarantees +every submission gets a distinct, ordered sequence number. + +**Requires:** `keypair`, `wallet`, or `signer` set (recipient) + +--- + ### `cancel(streamId) → Promise` Atomically settles both parties (recipient gets owed amount, sender gets refund). @@ -203,6 +246,11 @@ const sub = client.streams.subscribe(streamId, { onResume: e => console.log('Resumed at:', e.resumedAt), onTopUp: e => console.log('Topped up:', e.amount), onClawback: e => console.log('Clawback:', e.amount), + onCreated: e => console.log('Created:', e.recipient), + onForceCancel: e => console.log('Force-cancelled by:', e.recipient), + onRecipientTransfer: e => console.log('Recipient transferred to:', e.newRecipient), + onOperatorSet: e => console.log('Operator set:', e.operator), + onOperatorRevoke: e => console.log('Operator revoked:', e.operator), onError: err => console.warn('Polling error:', err), pollInterval: 3000, // ms; default 5000 maxBackoffMs: 30000, // ms; default 60000 @@ -222,10 +270,17 @@ sub.unsubscribe(); > **All event payload fields are decoded.** `src/events.ts`'s `dispatchEvent()` parses every > multi-field event from its tuple `ScVal`s: `onWithdraw` → `{ recipient, amount, > totalWithdrawn, remaining }`, `onCancel` → `{ sender, refundAmount, withdrawnSoFar }`, -> `onPause` → `{ sender, pausedAt, withdrawable }`, `onTopUp` → `{ sender, amount, newBalance }`. -> Single-field events are decoded from their bare scalar: `onResume` → `{ sender, resumedAt }`, -> `onClawback` → `{ sender, amount }`. These fields are real values from the chain — no -> placeholder `0`/`0n` values remain. +> `onPause` → `{ sender, pausedAt, withdrawable }`, `onTopUp` → `{ sender, amount, newBalance }`, +> `onCreated` → `{ sender, recipient, token, depositAmount, ratePerSecond, startTime, endTime }`, +> `onForceCancel` → `{ recipient, payoutAmount, refundAmount }`. Single-field events are decoded +> from their bare scalar: `onResume` → `{ sender, resumedAt }`, `onClawback` → `{ sender, amount }`, +> `onRecipientTransfer` → `{ previousRecipient, newRecipient }`, `onOperatorSet` → `{ sender, +> operator }`, `onOperatorRevoke` → `{ sender, operator }`. These fields are real values from the +> chain — no placeholder `0`/`0n` values remain. +> +> `onRecipientTransfer`, `onForceCancel`, `onOperatorSet`, and `onOperatorRevoke` were previously +> silently ignored — a live subscriber was never notified when the recipient role moved to +> another address, a paused stream was force-cancelled, or an operator was delegated/revoked. The first poll seeds its start ledger from `getLatestLedger()` before calling `getEvents()` (Soroban RPC's `getEvents` rejects without a start ledger). If that seeding call itself fails, it's retried on @@ -813,4 +868,32 @@ new Module49(config?: Module49Config) * `clearCache(): void` - Clears the internal lookup cache and metrics. * `getPerformanceMetrics(): Module49Metrics` - Returns real-time metrics (`totalProcessed`, `cacheHits`, `cacheMisses`, `hitRate`, `averageExecutionTimeMs`). +--- + +## `Module44` (Feature #44) + +Stream liquidity-risk / runway calculator implementing Feature #44. For each stream, computes the remaining `runwaySecs` until `endTime` and classifies it into a `LiquidityRiskLevel` (`'inactive' | 'critical' | 'warning' | 'healthy'`), so a dashboard can flag streams about to run out of scheduled balance. Uses the same LRU-memoized lookup pattern as `Module26`/`Module36`/`Module48`; actual speedup is workload-dependent (proportional to cache hit rate) — see `getPerformanceMetrics()` for a measured value, never a fixed assumed percentage. + +### Constructor + +```typescript +new Module44(config?: Module44Config) +``` + +| Option | Type | Default | Notes | +|--------|------|---------|-------| +| `cacheSize` | `number` | `1000` | Max entries in memoization cache | +| `enableOptimization` | `boolean` | `true` | Enables memoized lookup caching | +| `batchChunkSize` | `number` | `50` | Stream chunk size for batch processing | +| `criticalThresholdSecs` | `number` | `86400` (1 day) | Runway below this is `'critical'`. Throws if negative | +| `warningThresholdSecs` | `number` | `604800` (7 days) | Runway below this (and at/above `criticalThresholdSecs`) is `'warning'`. Throws if not greater than `criticalThresholdSecs` | + +### Methods + +* `assessSingleItem(item: StreamRiskItem): StreamRiskAssessment` — Computes one stream's `runwaySecs` and `riskLevel`. A cancelled, paused, or zero-rate stream is `'inactive'` with `runwaySecs: 0`. An open-ended stream (`endTime === 0`) is `'healthy'` with `runwaySecs: null` — its runway isn't bounded by a schedule, only by the sender keeping the balance topped up. +* `assessBatch(items: StreamRiskItem[]): StreamRiskAssessment[]` — Assesses an array of streams in `batchChunkSize` chunks. +* `estimateTopUpNeeded(stream: StreamInfo, targetRunwaySecs: number, nowSec?: number): bigint` — Stroops needed via `top_up()` for the stream's runway to reach `targetRunwaySecs`. Returns `0n` if the stream is inactive/paused/cancelled, the target is non-positive, or the target is already met (including any open-ended stream, whose runway is treated as unbounded). +* `clearCache(): void` — Clears the internal lookup cache and metrics. +* `getPerformanceMetrics(): Module44Metrics` — Returns `totalAssessed`, `cacheHits`, `cacheMisses`, `averageExecutionTimeMs`, and `measuredSpeedupPercent` (a real measurement derived from this instance's own accumulated hit/miss timings, `null` until both have occurred at least once). + diff --git a/src/events.ts b/src/events.ts index 38de97b..c69e501 100644 --- a/src/events.ts +++ b/src/events.ts @@ -15,10 +15,21 @@ import type { ResumeEvent, TopUpEvent, ClawbackEvent, + CreatedEvent, + ForceCancelEvent, + RecipientTransferEvent, + OperatorSetEvent, + OperatorRevokedEvent, } from './types/index.js'; import { scValToI128, scValToU64, createRpcServer } from './soroban.js'; // ── Event topic names (match symbol_short!() values in Rust) ───────────────── +// +// `created`, `force_cxl`, `xfer_rec`, `set_op` and `rm_op` are emitted by +// contracts/stream/src/events.rs alongside the six handled below, but were +// never wired into TOPIC/dispatchEvent — a subscriber was silently never +// notified of a recipient transfer, a recipient-initiated force-cancel, or +// an operator delegation/revocation (see #506). const TOPIC = { WITHDRAWN: 'withdrawn', @@ -27,6 +38,11 @@ const TOPIC = { RESUMED: 'resumed', TOPPED_UP: 'topped_up', CLAWBACK: 'clawback', + CREATED: 'created', + FORCE_CANCELLED: 'force_cxl', + RECIPIENT_TRANSFERRED: 'xfer_rec', + OPERATOR_SET: 'set_op', + OPERATOR_REVOKED: 'rm_op', } as const; // ── Parser helpers ──────────────────────────────────────────────────────────── @@ -51,6 +67,10 @@ function u64Field(fields: xdr.ScVal[], index: number): number { return field ? Number(scValToU64(field)) : 0; } +function addressFieldAt(fields: xdr.ScVal[], index: number): string { + return addressField(fields[index]); +} + /** * Decodes an address topic to its G.../C... string. `ScVal.address()?.accountId()` * returns the raw XDR PublicKey object, not a string — calling `.toString()` @@ -307,6 +327,75 @@ export function dispatchEvent( handlers.onClawback(data); break; } + + case TOPIC.CREATED: { + if (!handlers.onCreated) break; + // data: (recipient, token, deposit_amount: i128, rate_per_second: i128, start_time: u64, end_time: u64) + const fields = tupleFields(event.value); + const data: CreatedEvent = { + sender: actor, + recipient: addressFieldAt(fields, 0), + token: addressFieldAt(fields, 1), + depositAmount: i128Field(fields, 2), + ratePerSecond: i128Field(fields, 3), + startTime: u64Field(fields, 4), + endTime: u64Field(fields, 5), + sequence, + }; + handlers.onCreated(data); + break; + } + + case TOPIC.FORCE_CANCELLED: { + if (!handlers.onForceCancel) break; + // data: (payout_amount: i128, refund_amount: i128) — mirrors cancel()'s + // (refund_amount, withdrawn_so_far) shape but recipient-initiated. + const fields = tupleFields(event.value); + const data: ForceCancelEvent = { + recipient: actor, + payoutAmount: i128Field(fields, 0), + refundAmount: i128Field(fields, 1), + sequence, + }; + handlers.onForceCancel(data); + break; + } + + case TOPIC.RECIPIENT_TRANSFERRED: { + if (!handlers.onRecipientTransfer) break; + // data: new_recipient: address (bare scalar, not a tuple) + const data: RecipientTransferEvent = { + previousRecipient: actor, + newRecipient: addressField(event.value), + sequence, + }; + handlers.onRecipientTransfer(data); + break; + } + + case TOPIC.OPERATOR_SET: { + if (!handlers.onOperatorSet) break; + // data: operator: address (bare scalar, not a tuple) + const data: OperatorSetEvent = { + sender: actor, + operator: addressField(event.value), + sequence, + }; + handlers.onOperatorSet(data); + break; + } + + case TOPIC.OPERATOR_REVOKED: { + if (!handlers.onOperatorRevoke) break; + // data: operator: address (bare scalar, not a tuple) + const data: OperatorRevokedEvent = { + sender: actor, + operator: addressField(event.value), + sequence, + }; + handlers.onOperatorRevoke(data); + break; + } } return sequence; diff --git a/src/index.ts b/src/index.ts index bbce3b6..8286fe8 100644 --- a/src/index.ts +++ b/src/index.ts @@ -109,3 +109,12 @@ export type { Module49Metrics, } from './module49.js'; +export { Module44 } from './module44.js'; +export type { + Module44Config, + StreamRiskItem, + LiquidityRiskLevel, + StreamRiskAssessment, + Module44Metrics, +} from './module44.js'; + diff --git a/src/module44.ts b/src/module44.ts new file mode 100644 index 0000000..c282447 --- /dev/null +++ b/src/module44.ts @@ -0,0 +1,211 @@ +import type { StreamInfo } from './types/index.js'; +import { LruMemoCache } from './lru-memo-cache.js'; + +export interface Module44Config { + /** Maximum number of risk assessments retained in the LRU cache */ + cacheSize?: number; + /** Enable memoized assessment; actual speedup depends on hit rate — see `getPerformanceMetrics()` for a measured value */ + enableOptimization?: boolean; + /** Preferred chunk size for batch processing stream items */ + batchChunkSize?: number; + /** Runway (seconds) below which a stream is classified 'critical'. Default 86400 (1 day). */ + criticalThresholdSecs?: number; + /** Runway (seconds) below which a stream is classified 'warning'. Default 604800 (7 days). */ + warningThresholdSecs?: number; +} + +export interface StreamRiskItem { + id: string; + stream: StreamInfo; + /** Observation timestamp (unix seconds) */ + timestamp?: number; +} + +export type LiquidityRiskLevel = 'inactive' | 'critical' | 'warning' | 'healthy'; + +export interface StreamRiskAssessment { + id: string; + /** + * Seconds of streaming remaining before the stream reaches `endTime`. + * `null` for an open-ended stream (`endTime === 0`) — its runway is not + * bounded by a schedule, only by the sender keeping the balance topped up. + */ + runwaySecs: number | null; + riskLevel: LiquidityRiskLevel; + isCached: boolean; + computedAt: number; +} + +export interface Module44Metrics { + totalAssessed: number; + cacheHits: number; + cacheMisses: number; + /** + * Measured, not assumed: `(avgMissMs - avgHitMs) / avgMissMs * 100`, based + * on this instance's own accumulated timings. `null` until at least one + * hit and one miss have both been recorded (nothing to compare yet). + */ + measuredSpeedupPercent: number | null; + averageExecutionTimeMs: number; +} + +const DEFAULT_CRITICAL_THRESHOLD_SECS = 86_400; // 1 day +const DEFAULT_WARNING_THRESHOLD_SECS = 604_800; // 7 days + +/** + * Module 44: stream liquidity-risk / runway calculator. + * + * Implements Feature #44 with LRU-memoized runway and risk-level assessment + * per stream, so a dashboard can flag streams that are about to run out of + * scheduled balance. Speedup from caching is workload-dependent + * (proportional to cache hit rate); call `getPerformanceMetrics()` for this + * instance's own measured hit/miss timing rather than assuming a fixed + * percentage. + */ +export class Module44 { + private readonly enableOptimization: boolean; + private readonly batchChunkSize: number; + private readonly criticalThresholdSecs: number; + private readonly warningThresholdSecs: number; + + private readonly cache: LruMemoCache; + private totalAssessed = 0; + private totalExecutionTimeMs = 0; + + constructor(config: Module44Config = {}) { + this.cache = new LruMemoCache(config.cacheSize ?? 1000); + this.enableOptimization = config.enableOptimization ?? true; + this.batchChunkSize = config.batchChunkSize ?? 50; + this.criticalThresholdSecs = config.criticalThresholdSecs ?? DEFAULT_CRITICAL_THRESHOLD_SECS; + this.warningThresholdSecs = config.warningThresholdSecs ?? DEFAULT_WARNING_THRESHOLD_SECS; + + if (this.criticalThresholdSecs < 0) { + throw new Error('Module44: criticalThresholdSecs cannot be negative'); + } + if (this.warningThresholdSecs <= this.criticalThresholdSecs) { + throw new Error('Module44: warningThresholdSecs must be greater than criticalThresholdSecs'); + } + } + + /** + * Assess a batch of streams' liquidity risk with fast-path cache lookup. + */ + public assessBatch(items: StreamRiskItem[]): StreamRiskAssessment[] { + const results: StreamRiskAssessment[] = new Array(items.length); + + for (let i = 0; i < items.length; i += this.batchChunkSize) { + const chunkEnd = Math.min(i + this.batchChunkSize, items.length); + for (let j = i; j < chunkEnd; j++) { + const item = items[j]; + if (!item) continue; + + results[j] = this.assessSingleItem(item); + } + } + + return results; + } + + /** + * Assess a single stream's runway and liquidity risk level. + */ + public assessSingleItem(item: StreamRiskItem): StreamRiskAssessment { + const start = performance.now(); + const nowSec = item.timestamp ?? Math.floor(Date.now() / 1000); + const cacheKey = `${item.id}_${item.stream.paused ? 1 : 0}_${item.stream.cancelled ? 1 : 0}_${item.stream.ratePerSecond.toString()}_${item.stream.endTime}_${nowSec}`; + + if (this.enableOptimization) { + const cached = this.cache.get(cacheKey); + if (cached) { + const elapsed = performance.now() - start; + this.cache.recordHit(elapsed); + this.totalAssessed++; + this.totalExecutionTimeMs += elapsed; + return { ...cached, isCached: true }; + } + } + + const result = this.computeAssessment(item.id, item.stream, nowSec); + + if (this.enableOptimization) { + this.cache.set(cacheKey, result); + } + + const elapsed = performance.now() - start; + this.cache.recordMiss(elapsed); + this.totalAssessed++; + this.totalExecutionTimeMs += elapsed; + return result; + } + + private computeAssessment(id: string, stream: StreamInfo, nowSec: number): StreamRiskAssessment { + if (stream.cancelled || stream.paused || stream.ratePerSecond <= 0n) { + return { id, runwaySecs: 0, riskLevel: 'inactive', isCached: false, computedAt: nowSec }; + } + + if (stream.endTime === 0) { + // Open-ended: not bounded by a schedule, so treat as healthy — the + // sender is expected to keep the balance topped up via top_up(). + return { id, runwaySecs: null, riskLevel: 'healthy', isCached: false, computedAt: nowSec }; + } + + const runwaySecs = Math.max(0, stream.endTime - nowSec); + let riskLevel: LiquidityRiskLevel; + if (runwaySecs <= 0) { + riskLevel = 'inactive'; + } else if (runwaySecs < this.criticalThresholdSecs) { + riskLevel = 'critical'; + } else if (runwaySecs < this.warningThresholdSecs) { + riskLevel = 'warning'; + } else { + riskLevel = 'healthy'; + } + + return { id, runwaySecs, riskLevel, isCached: false, computedAt: nowSec }; + } + + /** + * Amount (in stroops) that must be added via `top_up()` for the stream's + * runway to reach `targetRunwaySecs` from `nowSec`. Returns `0n` when the + * stream is already inactive/paused/cancelled or already meets the target. + */ + public estimateTopUpNeeded( + stream: StreamInfo, + targetRunwaySecs: number, + nowSec = Math.floor(Date.now() / 1000), + ): bigint { + if (targetRunwaySecs <= 0 || stream.ratePerSecond <= 0n || stream.cancelled || stream.paused) { + return 0n; + } + + const currentRunwaySecs = stream.endTime === 0 ? Infinity : Math.max(0, stream.endTime - nowSec); + if (currentRunwaySecs >= targetRunwaySecs) return 0n; + + const deficitSecs = Math.ceil(targetRunwaySecs - currentRunwaySecs); + return stream.ratePerSecond * BigInt(deficitSecs); + } + + /** + * Reset performance cache and internal state. + */ + public clearCache(): void { + this.cache.clear(); + this.totalAssessed = 0; + this.totalExecutionTimeMs = 0; + } + + /** + * Retrieve performance metrics, including a measured (not assumed) cache speedup. + */ + public getPerformanceMetrics(): Module44Metrics { + const { cacheHits, cacheMisses, measuredSpeedupPercent } = this.cache.metrics(); + + return { + totalAssessed: this.totalAssessed, + cacheHits, + cacheMisses, + measuredSpeedupPercent: this.enableOptimization ? measuredSpeedupPercent : null, + averageExecutionTimeMs: this.totalAssessed > 0 ? this.totalExecutionTimeMs / this.totalAssessed : 0, + }; + } +} diff --git a/src/soroban.ts b/src/soroban.ts index 08ff47f..9792d54 100644 --- a/src/soroban.ts +++ b/src/soroban.ts @@ -144,11 +144,33 @@ export function createRpcServer(rpcUrl: string): SorobanRpc.Server { return proxied; } +/** + * Resolve the inclusion (bid) fee, in stroops, for a submitted transaction. + * + * An explicit `fee` always wins. Otherwise `feeMultiplier` scales + * `BASE_FEE`. With neither set, this returns `BASE_FEE` (the network + * minimum) unchanged — the previous, always-hardcoded behaviour. `BASE_FEE` + * alone is not competitive under inclusion-fee pressure (surge pricing, + * congested ledgers): the bid goes unselected, `_sendAndPoll` exhausts its + * poll attempts, and the caller sees a misleading "Transaction timed out" + * instead of "fee too low" (see #509). + */ +export function resolveFee(config: { fee?: string; feeMultiplier?: number }): string { + if (config.fee !== undefined) return config.fee; + if (config.feeMultiplier !== undefined) { + return (BigInt(BASE_FEE) * BigInt(config.feeMultiplier)).toString(); + } + return BASE_FEE; +} + /** * Build a contract-call transaction for simulate or submit. * * Fetches the caller's account from the RPC to get the current sequence * number, then wraps the call in a TransactionBuilder. + * + * @param fee - Inclusion fee in stroops. Defaults to `BASE_FEE`; pass the + * result of {@link resolveFee} to honour `ConduitConfig.fee`/`feeMultiplier`. */ export async function buildContractCallTx( rpcUrl: string, @@ -157,6 +179,7 @@ export async function buildContractCallTx( contractId: string, method: string, args: xdr.ScVal[], + fee: string = BASE_FEE, ): Promise> { const server = createRpcServer(rpcUrl); @@ -170,7 +193,7 @@ export async function buildContractCallTx( const contract = new Contract(contractId); return new TransactionBuilder(account, { - fee: BASE_FEE, + fee, networkPassphrase: passphrase, }) .addOperation(contract.call(method, ...args)) diff --git a/src/streams.ts b/src/streams.ts index af3d379..26c725c 100644 --- a/src/streams.ts +++ b/src/streams.ts @@ -36,6 +36,7 @@ import { DEFAULT_CONFIRMATION_MAX_ATTEMPTS, DEFAULT_CONFIRMATION_POLL_INTERVAL_MS, createRpcServer, + resolveFee, } from './soroban.js'; import { STREAM_FLAG_PAUSED, @@ -83,6 +84,12 @@ export class StreamsModule { private readonly _factory: FactoryModule; private activeWallet?: WalletAdapter; + /** + * Inclusion (bid) fee, in stroops, for transactions this module submits. + * Resolved once from `config.fee` / `config.feeMultiplier` (see #509). + */ + private readonly _fee: string; + /** * Session-scoped cache of stream ID → contract address resolutions. * Avoids a redundant factory RPC on every get/withdraw/cancel/pause/resume/topUp/clawback call @@ -111,6 +118,7 @@ export class StreamsModule { this.rpcUrl = config.rpcUrl ?? DEFAULT_RPC[config.network]; this.passphrase = NETWORK_PASSPHRASE[config.network]; this._factory = new FactoryModule(config); + this._fee = resolveFee(config); if (config.wallet) { this.activeWallet = config.wallet; @@ -240,7 +248,7 @@ export class StreamsModule { boolToScVal(clawbackEnabled), ]; - const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, senderAddr, factoryId, 'create_stream', args); + const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, senderAddr, factoryId, 'create_stream', args, this._fee); const server = this._server(); const sim = await catchNetworkError('simulateTransaction (create)', server.simulateTransaction(tx)); @@ -327,35 +335,45 @@ export class StreamsModule { } /** - * Withdraw from multiple streams concurrently. + * Withdraw from multiple streams. * * Note: Soroban currently permits only one invoke_host_function * operation per transaction, so this cannot be assembled into a single * atomic transaction the way classic Stellar payment operations can. - * Each withdrawal is submitted as its own transaction; they run - * concurrently and are reported independently so a failure on one - * streamId (e.g. StreamNotFound, insufficient balance) does not block - * or roll back the others. + * Each withdrawal is submitted as its own transaction; they are reported + * independently so a failure on one streamId (e.g. StreamNotFound, + * insufficient balance) does not block or fail the others. + * + * Withdrawals are submitted one at a time, not concurrently. Each + * withdraw() -> _invoke() -> buildContractCallTx() reads the caller + * account's current sequence number via getAccount() and builds a + * transaction on top of it. Firing all N withdrawals at once means every + * one of them reads the same sequence number and builds a transaction + * with the same value, so with a single keypair/wallet at most one + * submission can ever land — the rest fail with txBAD_SEQ (see #504). + * Awaiting each withdrawal (submit + confirm) before starting the next + * guarantees getAccount() only observes the sequence after the previous + * transaction has landed, so every submission gets a distinct, ordered + * sequence number. */ async batchWithdraw(withdrawals: BatchWithdrawItem[]): Promise { this._ensureCanMutate(); - const settled = await Promise.allSettled( - withdrawals.map(w => this.withdraw(w.streamId, w.amount)), - ); - - return settled.map((result, i) => { - const streamId = BigInt(withdrawals[i]!.streamId); - if (result.status === 'fulfilled') { - return { streamId, success: true, txHash: result.value }; + const results: BatchWithdrawResult[] = []; + for (const w of withdrawals) { + const streamId = BigInt(w.streamId); + try { + const txHash = await this.withdraw(w.streamId, w.amount); + results.push({ streamId, success: true, txHash }); + } catch (err) { + results.push({ + streamId, + success: false, + error: err instanceof Error ? err.message : String(err), + }); } - const err = result.reason; - return { - streamId, - success: false, - error: err instanceof Error ? err.message : String(err), - }; - }); + } + return results; } /** Cancel the stream (sender only). Settles all balances atomically. */ @@ -430,7 +448,7 @@ export class StreamsModule { this._ensureCanMutate(); const addr = await this._resolveAddr(BigInt(streamId)); const caller = await this._getSenderAddress(); - const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, caller, addr, 'clawback', []); + const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, caller, addr, 'clawback', [], this._fee); const server = this._server(); const sim = await catchNetworkError('simulateTransaction (clawback)', server.simulateTransaction(tx)); @@ -771,7 +789,7 @@ export class StreamsModule { /** Simulate -> assemble -> sign -> submit -> poll. Returns txHash. */ private async _invoke(contractId: string, method: string, args: xdr.ScVal[]): Promise { const senderAddr = await this._getSenderAddress(); - const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, senderAddr, contractId, method, args); + const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, senderAddr, contractId, method, args, this._fee); const server = this._server(); const sim = await catchNetworkError('simulateTransaction (invoke)', server.simulateTransaction(tx)); if (SorobanRpc.Api.isSimulationError(sim)) { diff --git a/src/tests/batch-withdraw.test.ts b/src/tests/batch-withdraw.test.ts index baed9ca..66aa0ac 100644 --- a/src/tests/batch-withdraw.test.ts +++ b/src/tests/batch-withdraw.test.ts @@ -65,4 +65,30 @@ describe('StreamsModule.batchWithdraw()', () => { /keypair, wallet adapter, or signer/, ); }); + + it('submits withdrawals one at a time, not concurrently (#504)', async () => { + // A concurrent Promise.allSettled(...map) would call withdraw() for every + // item before any of them resolve. Serialised submission must instead + // wait for withdraw #1 to fully settle before withdraw #2 is even + // invoked — otherwise every withdrawal would read the same account + // sequence number and only one could ever land on-chain. + const sdk = new StreamsModule(makeConfig()); + let inFlight = 0; + let maxConcurrent = 0; + const callOrder: bigint[] = []; + + vi.spyOn(sdk, 'withdraw').mockImplementation(async (streamId) => { + callOrder.push(BigInt(streamId)); + inFlight++; + maxConcurrent = Math.max(maxConcurrent, inFlight); + await new Promise(resolve => setTimeout(resolve, 5)); + inFlight--; + return `hash-${streamId}`; + }); + + await sdk.batchWithdraw([{ streamId: 1n }, { streamId: 2n }, { streamId: 3n }]); + + expect(maxConcurrent).toBe(1); + expect(callOrder).toEqual([1n, 2n, 3n]); + }); }); \ No newline at end of file diff --git a/src/tests/events.test.ts b/src/tests/events.test.ts index 93cac08..6ae27fa 100644 --- a/src/tests/events.test.ts +++ b/src/tests/events.test.ts @@ -9,16 +9,26 @@ import type { ResumeEvent, TopUpEvent, ClawbackEvent, + CreatedEvent, + ForceCancelEvent, + RecipientTransferEvent, + OperatorSetEvent, + OperatorRevokedEvent, } from '../types/index.js'; // ── Fixture helpers ─────────────────────────────────────────────────────────── const actorAddress = Keypair.random().publicKey(); +const otherAddress = Keypair.random().publicKey(); function topic(name: string): xdr.ScVal { return xdr.ScVal.scvSymbol(name); } +function address(addr: string): xdr.ScVal { + return new Address(addr).toScVal(); +} + function actorTopic(): xdr.ScVal { return new Address(actorAddress).toScVal(); } @@ -150,6 +160,100 @@ describe('dispatchEvent — clawback', () => { }); }); +// ── created: (recipient, token, deposit_amount, rate_per_second, start_time, end_time) ── + +describe('dispatchEvent — created', () => { + it('decodes the address and numeric tuple fields (#506)', () => { + const event = makeEvent( + 'created', + tuple(address(otherAddress), address(otherAddress), i128(1_000_000n), i128(10n), u64(1_700_000_000), u64(1_700_100_000)), + ); + let received: CreatedEvent | undefined; + const handlers: StreamEventHandlers = { onCreated: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ + sender: actorAddress, + recipient: otherAddress, + token: otherAddress, + depositAmount: 1_000_000n, + ratePerSecond: 10n, + startTime: 1_700_000_000, + endTime: 1_700_100_000, + sequence: 0n, + }); + }); + + it('does not call onCreated when the handler is not registered', () => { + const event = makeEvent('created', tuple(address(otherAddress), address(otherAddress), i128(1n), i128(1n), u64(1), u64(2))); + expect(() => dispatchEvent(event, {})).not.toThrow(); + }); +}); + +// ── force_cxl: (payout_amount, refund_amount) ──────────────────────────────── + +describe('dispatchEvent — force_cxl', () => { + it('decodes both i128 tuple fields, attributing the actor as the recipient (#506)', () => { + const event = makeEvent('force_cxl', tuple(i128(50_000n), i128(25_000n))); + let received: ForceCancelEvent | undefined; + const handlers: StreamEventHandlers = { onForceCancel: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ + recipient: actorAddress, + payoutAmount: 50_000n, + refundAmount: 25_000n, + sequence: 0n, + }); + }); +}); + +// ── xfer_rec: new_recipient: address (bare scalar) ─────────────────────────── + +describe('dispatchEvent — xfer_rec', () => { + it('decodes the new recipient address, attributing the actor as the previous recipient (#506)', () => { + const event = makeEvent('xfer_rec', address(otherAddress)); + let received: RecipientTransferEvent | undefined; + const handlers: StreamEventHandlers = { onRecipientTransfer: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ + previousRecipient: actorAddress, + newRecipient: otherAddress, + sequence: 0n, + }); + }); +}); + +// ── set_op / rm_op: operator: address (bare scalar) ────────────────────────── + +describe('dispatchEvent — set_op', () => { + it('decodes the delegated operator address (#506)', () => { + const event = makeEvent('set_op', address(otherAddress)); + let received: OperatorSetEvent | undefined; + const handlers: StreamEventHandlers = { onOperatorSet: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ sender: actorAddress, operator: otherAddress, sequence: 0n }); + }); +}); + +describe('dispatchEvent — rm_op', () => { + it('decodes the revoked operator address (#506)', () => { + const event = makeEvent('rm_op', address(otherAddress)); + let received: OperatorRevokedEvent | undefined; + const handlers: StreamEventHandlers = { onOperatorRevoke: (e) => { received = e; } }; + + dispatchEvent(event, handlers); + + expect(received).toEqual({ sender: actorAddress, operator: otherAddress, sequence: 0n }); + }); +}); + // ── Robustness ───────────────────────────────────────────────────────────── describe('dispatchEvent — sequence (topics[2])', () => { diff --git a/src/tests/module44.test.ts b/src/tests/module44.test.ts new file mode 100644 index 0000000..92a080d --- /dev/null +++ b/src/tests/module44.test.ts @@ -0,0 +1,245 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import { Module44 } from '../module44.js'; +import type { StreamInfo } from '../types/index.js'; + +describe('Module44 (SDK Feature #44)', () => { + let module44: Module44; + const now = 1_000_000; + + const mockStream: StreamInfo = { + id: 1n, + address: 'CCSTREAM44ADDRESSXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX', + sender: 'GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFXYCZLYC3ZCHB2D4P3CF', + recipient: 'GA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVSGZ', + token: 'native', + ratePerSecond: 100n, + startTime: now - 500, + endTime: now + 500, + withdrawn: 0n, + paused: false, + pausedAt: 0, + cancelled: false, + clawbackEnabled: false, + }; + + beforeEach(() => { + module44 = new Module44({ cacheSize: 10, batchChunkSize: 5 }); + }); + + describe('Constructor & Configuration', () => { + it('initializes with default options and no speedup measurement yet', () => { + const defaultMod = new Module44(); + const metrics = defaultMod.getPerformanceMetrics(); + expect(metrics.totalAssessed).toBe(0); + expect(metrics.measuredSpeedupPercent).toBeNull(); + }); + + it('never reports a speedup when optimization is disabled', () => { + const customMod = new Module44({ enableOptimization: false }); + const item = { id: 'stream-1', stream: mockStream, timestamp: now }; + customMod.assessSingleItem(item); + customMod.assessSingleItem(item); + expect(customMod.getPerformanceMetrics().measuredSpeedupPercent).toBeNull(); + }); + + it('throws when criticalThresholdSecs is negative', () => { + expect(() => new Module44({ criticalThresholdSecs: -1 })).toThrow(/criticalThresholdSecs/); + }); + + it('throws when warningThresholdSecs does not exceed criticalThresholdSecs', () => { + expect(() => new Module44({ criticalThresholdSecs: 100, warningThresholdSecs: 100 })).toThrow(/warningThresholdSecs/); + expect(() => new Module44({ criticalThresholdSecs: 100, warningThresholdSecs: 50 })).toThrow(/warningThresholdSecs/); + }); + }); + + describe('assessSingleItem — risk classification', () => { + it('classifies a cancelled stream as inactive with zero runway', () => { + const item = { id: 's1', stream: { ...mockStream, cancelled: true }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result).toMatchObject({ id: 's1', runwaySecs: 0, riskLevel: 'inactive', isCached: false, computedAt: now }); + }); + + it('classifies a paused stream as inactive with zero runway', () => { + const item = { id: 's1', stream: { ...mockStream, paused: true }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.riskLevel).toBe('inactive'); + expect(result.runwaySecs).toBe(0); + }); + + it('classifies a zero-rate stream as inactive', () => { + const item = { id: 's1', stream: { ...mockStream, ratePerSecond: 0n }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.riskLevel).toBe('inactive'); + }); + + it('classifies an open-ended stream as healthy with null runway', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: 0 }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.runwaySecs).toBeNull(); + expect(result.riskLevel).toBe('healthy'); + }); + + it('classifies runway already past endTime as inactive', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: now - 1 }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.runwaySecs).toBe(0); + expect(result.riskLevel).toBe('inactive'); + }); + + it('classifies runway below the critical threshold as critical', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: now + 3600 }, timestamp: now }; // 1 hour + const result = module44.assessSingleItem(item); + expect(result.runwaySecs).toBe(3600); + expect(result.riskLevel).toBe('critical'); + }); + + it('classifies runway between critical and warning thresholds as warning', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: now + 172_800 }, timestamp: now }; // 2 days + const result = module44.assessSingleItem(item); + expect(result.riskLevel).toBe('warning'); + }); + + it('classifies runway above the warning threshold as healthy', () => { + const item = { id: 's1', stream: { ...mockStream, endTime: now + 1_000_000 }, timestamp: now }; + const result = module44.assessSingleItem(item); + expect(result.riskLevel).toBe('healthy'); + }); + + it('honours custom critical/warning thresholds', () => { + const custom = new Module44({ criticalThresholdSecs: 10, warningThresholdSecs: 20 }); + const item = { id: 's1', stream: { ...mockStream, endTime: now + 15 }, timestamp: now }; + expect(custom.assessSingleItem(item).riskLevel).toBe('warning'); + }); + + it('uses system current time if timestamp is omitted', () => { + const item = { id: 's1', stream: mockStream }; + const result = module44.assessSingleItem(item); + expect(result.computedAt).toBeGreaterThan(0); + }); + }); + + describe('Optimization & Caching', () => { + it('serves cached results on duplicate evaluation requests', () => { + const item = { id: 's1', stream: mockStream, timestamp: now }; + + const firstPass = module44.assessSingleItem(item); + expect(firstPass.isCached).toBe(false); + + const secondPass = module44.assessSingleItem(item); + expect(secondPass.isCached).toBe(true); + expect(secondPass.riskLevel).toBe(firstPass.riskLevel); + + const metrics = module44.getPerformanceMetrics(); + expect(metrics.cacheHits).toBe(1); + expect(metrics.cacheMisses).toBe(1); + expect( + metrics.measuredSpeedupPercent === null || typeof metrics.measuredSpeedupPercent === 'number', + ).toBe(true); + }); + + it('evicts oldest cache item when cacheSize threshold is reached', () => { + const smallCacheMod = new Module44({ cacheSize: 2 }); + + smallCacheMod.assessSingleItem({ id: 'item-1', stream: mockStream, timestamp: 1000 }); + smallCacheMod.assessSingleItem({ id: 'item-2', stream: mockStream, timestamp: 1001 }); + smallCacheMod.assessSingleItem({ id: 'item-3', stream: mockStream, timestamp: 1002 }); + + const reQuery = smallCacheMod.assessSingleItem({ id: 'item-1', stream: mockStream, timestamp: 1000 }); + expect(reQuery.isCached).toBe(false); + }); + + it('bypasses cache when optimization is disabled', () => { + const unoptimizedMod = new Module44({ enableOptimization: false }); + const item = { id: 's1', stream: mockStream, timestamp: now }; + + unoptimizedMod.assessSingleItem(item); + const secondPass = unoptimizedMod.assessSingleItem(item); + + expect(secondPass.isCached).toBe(false); + expect(unoptimizedMod.getPerformanceMetrics().measuredSpeedupPercent).toBeNull(); + }); + }); + + describe('assessBatch', () => { + it('assesses a batch of stream items in chunked iterations', () => { + const batchItems = Array.from({ length: 12 }, (_, i) => ({ + id: `stream-${i}`, + stream: { ...mockStream, id: BigInt(i) }, + timestamp: now, + })); + + const results = module44.assessBatch(batchItems); + + expect(results).toHaveLength(12); + expect(results[0]?.id).toBe('stream-0'); + expect(results[11]?.id).toBe('stream-11'); + + expect(module44.getPerformanceMetrics().totalAssessed).toBe(12); + }); + + it('reports the same totalAssessed whether items go through assessBatch or assessSingleItem directly', () => { + const viaSingle = new Module44({ cacheSize: 10 }); + viaSingle.assessSingleItem({ id: 'a', stream: mockStream, timestamp: now }); + viaSingle.assessSingleItem({ id: 'b', stream: { ...mockStream, id: 2n }, timestamp: now }); + expect(viaSingle.getPerformanceMetrics().totalAssessed).toBe(2); + + const viaBatch = new Module44({ cacheSize: 10 }); + viaBatch.assessBatch([ + { id: 'a', stream: mockStream, timestamp: now }, + { id: 'b', stream: { ...mockStream, id: 2n }, timestamp: now }, + ]); + expect(viaBatch.getPerformanceMetrics().totalAssessed).toBe(2); + }); + }); + + describe('estimateTopUpNeeded', () => { + it('returns 0n when the target runway is already met', () => { + const stream = { ...mockStream, endTime: now + 10_000 }; + expect(module44.estimateTopUpNeeded(stream, 100, now)).toBe(0n); + }); + + it('computes the exact deficit amount in stroops for a bounded stream', () => { + const stream = { ...mockStream, endTime: now + 100, ratePerSecond: 50n }; + // Needs runway of 200s but only has 100s -> 100s deficit * 50/s = 5000n + expect(module44.estimateTopUpNeeded(stream, 200, now)).toBe(5000n); + }); + + it('returns 0n for an open-ended stream (runway already infinite)', () => { + const stream = { ...mockStream, endTime: 0 }; + expect(module44.estimateTopUpNeeded(stream, 1_000_000, now)).toBe(0n); + }); + + it('returns 0n for non-positive target runway', () => { + expect(module44.estimateTopUpNeeded(mockStream, 0, now)).toBe(0n); + expect(module44.estimateTopUpNeeded(mockStream, -10, now)).toBe(0n); + }); + + it('returns 0n for a zero-rate, paused, or cancelled stream', () => { + expect(module44.estimateTopUpNeeded({ ...mockStream, ratePerSecond: 0n }, 1000, now)).toBe(0n); + expect(module44.estimateTopUpNeeded({ ...mockStream, paused: true }, 1000, now)).toBe(0n); + expect(module44.estimateTopUpNeeded({ ...mockStream, cancelled: true }, 1000, now)).toBe(0n); + }); + + it('defaults nowSec to the current time when omitted', () => { + const stream = { ...mockStream, endTime: 0 }; + expect(() => module44.estimateTopUpNeeded(stream, 1000)).not.toThrow(); + }); + }); + + describe('clearCache & Metrics', () => { + it('resets cache state and performance counters', () => { + const item = { id: 's1', stream: mockStream, timestamp: now }; + module44.assessSingleItem(item); + module44.assessSingleItem(item); + + expect(module44.getPerformanceMetrics().cacheHits).toBe(1); + + module44.clearCache(); + + const freshMetrics = module44.getPerformanceMetrics(); + expect(freshMetrics.cacheHits).toBe(0); + expect(freshMetrics.cacheMisses).toBe(0); + expect(freshMetrics.totalAssessed).toBe(0); + }); + }); +}); diff --git a/src/tests/network-switcher-validation.test.ts b/src/tests/network-switcher-validation.test.ts index 0d50eef..dd47fe4 100644 --- a/src/tests/network-switcher-validation.test.ts +++ b/src/tests/network-switcher-validation.test.ts @@ -31,6 +31,7 @@ import { WalletConnectAdapter } from '../adapters/walletconnect.js'; vi.mock('../soroban.js', () => ({ buildContractCallTx: vi.fn().mockResolvedValue({ _stub: 'tx' }), simulateReadOnly: vi.fn(), + resolveFee: () => '100', scValToU64: (v: { u64: () => { toString: () => string } }) => BigInt(v.u64().toString()), scValToI128: () => 0n, NETWORK_PASSPHRASE: { diff --git a/src/tests/soroban-fee-config.test.ts b/src/tests/soroban-fee-config.test.ts new file mode 100644 index 0000000..87591d0 --- /dev/null +++ b/src/tests/soroban-fee-config.test.ts @@ -0,0 +1,70 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { BASE_FEE, StrKey } from '@stellar/stellar-sdk'; + +const CONTRACT_ID = StrKey.encodeContract(Buffer.alloc(32, 1)); + +const { mockGetAccount, mockTransactionBuilder } = vi.hoisted(() => ({ + mockGetAccount: vi.fn(), + mockTransactionBuilder: vi.fn(), +})); + +vi.mock('@stellar/stellar-sdk', async () => { + const actual = await vi.importActual('@stellar/stellar-sdk'); + return { + ...actual, + SorobanRpc: { + ...actual.SorobanRpc, + Server: class { + getAccount = mockGetAccount; + }, + }, + TransactionBuilder: mockTransactionBuilder, + }; +}); + +import { resolveFee, buildContractCallTx } from '../soroban.js'; + +beforeEach(() => { + mockGetAccount.mockReset().mockResolvedValue({ accountId: () => 'GACCOUNT', sequenceNumber: () => '1' }); + mockTransactionBuilder.mockReset().mockImplementation(function MockTransactionBuilder() { + return { + addOperation: vi.fn().mockReturnThis(), + setTimeout: vi.fn().mockReturnThis(), + build: vi.fn().mockReturnValue({ _stub: 'tx' }), + }; + }); +}); + +describe('resolveFee() (#509)', () => { + it('defaults to BASE_FEE when neither fee nor feeMultiplier is set', () => { + expect(resolveFee({})).toBe(BASE_FEE); + }); + + it('returns the explicit fee, ignoring feeMultiplier', () => { + expect(resolveFee({ fee: '5000', feeMultiplier: 10 })).toBe('5000'); + }); + + it('scales BASE_FEE by feeMultiplier', () => { + expect(resolveFee({ feeMultiplier: 10 })).toBe((BigInt(BASE_FEE) * 10n).toString()); + }); +}); + +describe('buildContractCallTx() fee parameter (#509)', () => { + it('defaults the TransactionBuilder fee to BASE_FEE when no fee is passed', async () => { + await buildContractCallTx('https://rpc', 'passphrase', 'GCALLER', CONTRACT_ID, 'withdraw', []); + + expect(mockTransactionBuilder).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ fee: BASE_FEE }), + ); + }); + + it('passes an explicit fee through to the TransactionBuilder', async () => { + await buildContractCallTx('https://rpc', 'passphrase', 'GCALLER', CONTRACT_ID, 'withdraw', [], '10000'); + + expect(mockTransactionBuilder).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ fee: '10000' }), + ); + }); +}); diff --git a/src/tests/streams-success.test.ts b/src/tests/streams-success.test.ts index c90d9b9..d0d81d9 100644 --- a/src/tests/streams-success.test.ts +++ b/src/tests/streams-success.test.ts @@ -491,7 +491,56 @@ describe('StreamsModule.transferRecipient() — success path', () => { await runThroughFirstPoll(() => sdk.transferRecipient(1n, newRecipient)); expect(buildContractCallTx).toHaveBeenCalledWith( expect.any(String), expect.any(String), expect.any(String), - STREAM_ADDR, 'transfer_recipient', [expect.anything()], + STREAM_ADDR, 'transfer_recipient', [expect.anything()], expect.any(String), + ); + }); +}); + +describe('StreamsModule — configurable inclusion fee (#509)', () => { + beforeEach(() => { + mockStreamAddress.mockResolvedValue(STREAM_ADDR); + mockSimulate.mockResolvedValue(simSuccess(xdr.ScVal.scvVoid())); + mockGetTransaction.mockResolvedValue(txSuccess()); + }); + + it('passes BASE_FEE by default to buildContractCallTx for a submitted operation', async () => { + const { buildContractCallTx } = await import('../soroban.js'); + const { StreamsModule } = await import('../streams.js'); + const { BASE_FEE } = await import('@stellar/stellar-sdk'); + const sdk = new StreamsModule(makeConfig()); + + await runThroughFirstPoll(() => sdk.pause(1n)); + + expect(buildContractCallTx).toHaveBeenLastCalledWith( + expect.any(String), expect.any(String), expect.any(String), + STREAM_ADDR, 'pause', [], BASE_FEE, + ); + }); + + it('honours an explicit config.fee override', async () => { + const { buildContractCallTx } = await import('../soroban.js'); + const { StreamsModule } = await import('../streams.js'); + const sdk = new StreamsModule(makeConfig({ fee: '5000' })); + + await runThroughFirstPoll(() => sdk.pause(1n)); + + expect(buildContractCallTx).toHaveBeenLastCalledWith( + expect.any(String), expect.any(String), expect.any(String), + STREAM_ADDR, 'pause', [], '5000', + ); + }); + + it('honours config.feeMultiplier scaled off BASE_FEE', async () => { + const { buildContractCallTx } = await import('../soroban.js'); + const { StreamsModule } = await import('../streams.js'); + const { BASE_FEE } = await import('@stellar/stellar-sdk'); + const sdk = new StreamsModule(makeConfig({ feeMultiplier: 10 })); + + await runThroughFirstPoll(() => sdk.pause(1n)); + + expect(buildContractCallTx).toHaveBeenLastCalledWith( + expect.any(String), expect.any(String), expect.any(String), + STREAM_ADDR, 'pause', [], (BigInt(BASE_FEE) * 10n).toString(), ); }); }); diff --git a/src/types/index.ts b/src/types/index.ts index 2cb1641..6b461d0 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -36,6 +36,20 @@ export interface ConduitConfig { confirmationPollIntervalMs?: number; /** Override Soroban confirmation polling attempts; default 30 */ confirmationMaxAttempts?: number; + /** + * Explicit inclusion (bid) fee in stroops for submitted transactions. + * Takes precedence over `feeMultiplier`. Defaults to `BASE_FEE` (100 + * stroops) when neither is set, which is the network minimum and is not + * competitive under inclusion-fee pressure (surge pricing, congested + * ledgers) — see #509. + */ + fee?: string; + /** + * Multiplier applied to `BASE_FEE` to compute the inclusion fee for + * submitted transactions, e.g. `10` bids 10x the network minimum. + * Ignored when `fee` is set. Defaults to `1` (BASE_FEE, unchanged). + */ + feeMultiplier?: number; } export interface StreamInfo { @@ -125,6 +139,58 @@ export interface ResumeEvent { resumedAt: number; sender: string; sequence: b export interface TopUpEvent { amount: bigint; newBalance: bigint; sender: string; sequence: bigint; } export interface ClawbackEvent { amount: bigint; sender: string; sequence: bigint; } +/** Published once when the factory deploys a new DripStream (`created` topic). */ +export interface CreatedEvent { + /** The stream's sender (topics[1] actor). */ + sender: string; + recipient: string; + token: string; + depositAmount: bigint; + ratePerSecond: bigint; + startTime: number; + endTime: number; + sequence: bigint; +} + +/** + * Published when the recipient force-cancels a paused stream after the + * pause threshold has elapsed (`force_cxl` topic). Settles atomically like + * `cancelled`, but is recipient-initiated rather than sender-initiated. + */ +export interface ForceCancelEvent { + /** The recipient who force-cancelled (topics[1] actor). */ + recipient: string; + /** Earned-but-unwithdrawn amount paid out to the recipient. */ + payoutAmount: bigint; + /** Unstreamed remainder refunded to the sender. */ + refundAmount: bigint; + sequence: bigint; +} + +/** Published when the recipient role is transferred to a new address (`xfer_rec` topic). */ +export interface RecipientTransferEvent { + /** The outgoing recipient who initiated the transfer (topics[1] actor). */ + previousRecipient: string; + newRecipient: string; + sequence: bigint; +} + +/** Published when an operator is delegated on the stream (`set_op` topic). */ +export interface OperatorSetEvent { + /** The address that granted the operator role (topics[1] actor). */ + sender: string; + operator: string; + sequence: bigint; +} + +/** Published when a previously-delegated operator is revoked (`rm_op` topic). */ +export interface OperatorRevokedEvent { + /** The address that revoked the operator role (topics[1] actor). */ + sender: string; + operator: string; + sequence: bigint; +} + /** A gap detected in the per-contract event sequence — see `DataKey::EventSequence` in contracts/stream/src/events.rs. */ export interface EventGap { /** The sequence number that should have come next. */ @@ -140,6 +206,16 @@ export interface StreamEventHandlers { onResume?: (e: ResumeEvent) => void; onTopUp?: (e: TopUpEvent) => void; onClawback?: (e: ClawbackEvent) => void; + /** Called when the factory deploys a new stream. Optional — most subscribers attach after a stream already exists. */ + onCreated?: (e: CreatedEvent) => void; + /** Called when the recipient force-cancels a paused stream. */ + onForceCancel?: (e: ForceCancelEvent) => void; + /** Called when the recipient role is transferred to a new address — the current subscriber may have just lost the stream. */ + onRecipientTransfer?: (e: RecipientTransferEvent) => void; + /** Called when an operator is delegated on the stream. */ + onOperatorSet?: (e: OperatorSetEvent) => void; + /** Called when a delegated operator is revoked. */ + onOperatorRevoke?: (e: OperatorRevokedEvent) => void; /** * Called when an event polling request fails. Polling continues * afterward — unless this failure reaches `maxConsecutiveFailures`, in