Skip to content

feat: add optional AbortSignal to every public async method #673

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ZacLou wants to merge 2 commits into
base: main
Choose a base branch
from
Open

feat: add optional AbortSignal to every public async method #673

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
320a6b6
feat: add optional AbortSignal to every public async method
ZacLou Sep 5, 2026
da24857
fix(streams): use AbortSignal in _resolveAddr, _simulateTx and _invoke
web-flow Sep 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 10 additions & 5 deletions src/factory.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,7 +131,8 @@ export class FactoryModule {
}

/** Total number of streams ever created through this factory. */
async streamCount(): Promise<bigint> {
async streamCount(signal?: AbortSignal): Promise<bigint> {
if (signal?.aborted) throw new DOMException("Aborted", "AbortError");
const caller = await this._resolveCallerAddress();
const tx = await buildContractCallTx(
this.rpcUrl, this.passphrase, caller,
Expand All @@ -142,9 +143,10 @@ export class FactoryModule {
}

/** Resolve a stream ID to its deployed contract address. Returns null if not found. */
async streamAddress(streamId: bigint | string): Promise<string | null> {
async streamAddress(streamId: bigint | string, signal?: AbortSignal): Promise<string | null> {
const id = BigInt(streamId);
const key = id.toString();
if (signal?.aborted) throw new DOMException("Aborted", "AbortError");

const cached = this.addressCache.get(key);
if (cached !== undefined) {
Expand Down Expand Up @@ -190,7 +192,8 @@ export class FactoryModule {
* enforce this itself, so an out-of-range value is silently clamped rather
* than sent through as-is (see #489).
*/
async streamsBySender(address: string, offset = 0, limit = DEFAULT_LIST_LIMIT): Promise<bigint[]> {
async streamsBySender(address: string, offset = 0, limit = DEFAULT_LIST_LIMIT, signal?: AbortSignal): Promise<bigint[]> {
if (signal?.aborted) throw new DOMException("Aborted", "AbortError");
const caller = await this._resolveCallerAddress();
const tx = await buildContractCallTx(
this.rpcUrl, this.passphrase, caller,
Expand All @@ -211,7 +214,8 @@ export class FactoryModule {
* enforce this itself, so an out-of-range value is silently clamped rather
* than sent through as-is (see #489).
*/
async streamsByRecipient(address: string, offset = 0, limit = DEFAULT_LIST_LIMIT): Promise<bigint[]> {
async streamsByRecipient(address: string, offset = 0, limit = DEFAULT_LIST_LIMIT, signal?: AbortSignal): Promise<bigint[]> {
if (signal?.aborted) throw new DOMException("Aborted", "AbortError");
const caller = await this._resolveCallerAddress();
const tx = await buildContractCallTx(
this.rpcUrl, this.passphrase, caller,
Expand All @@ -227,7 +231,8 @@ export class FactoryModule {
}

/** Current protocol fee in basis points (e.g. 30 = 0.3%). */
async protocolFeeBps(): Promise<number> {
async protocolFeeBps(signal?: AbortSignal): Promise<number> {
if (signal?.aborted) throw new DOMException("Aborted", "AbortError");
const caller = await this._resolveCallerAddress();
const tx = await buildContractCallTx(
this.rpcUrl, this.passphrase, caller,
Expand Down
58 changes: 37 additions & 21 deletions src/streams.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
* StreamsModule - all DripStream + DripFactory operations.
*/

import { SorobanRpc, nativeToScVal, xdr, Address, Transaction, BASE_FEE, Asset } from '@stellar/stellar-sdk';

Check warning on line 5 in src/streams.ts

View workflow job for this annotation

GitHub Actions / Lint & typecheck 

'SorobanRpc' import from '@stellar/stellar-sdk' is restricted. SorobanRpc is deprecated in @stellar/stellar-sdk v12. Import rpc instead
import type {
ConduitConfig,
CreateStreamParams,
Expand Down Expand Up @@ -316,8 +316,9 @@
}

/** Fetch full stream state from the deployed DripStream contract. */
async get(streamId: bigint | string): Promise<StreamInfo> {
async get(streamId: bigint | string, signal?: AbortSignal): Promise<StreamInfo> {
const id = BigInt(streamId);
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
const addr = await this._resolveAddr(id);
const caller = await this._resolveCallerAddress();
const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, caller, addr, 'info', []);
Expand Down Expand Up @@ -356,8 +357,9 @@
}

/** Get withdrawable balance - read-only, no transaction. */
async withdrawable(streamId: bigint | string): Promise<bigint> {
async withdrawable(streamId: bigint | string, signal?: AbortSignal): Promise<bigint> {
const id = BigInt(streamId);
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
const addr = await this._resolveAddr(id);
const caller = await this._resolveCallerAddress();
const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, caller, addr, 'withdrawable', []);
Expand All @@ -373,8 +375,9 @@
* useful for progress displays that shouldn't reset visually after a
* withdrawal. Read-only, no transaction.
*/
async streamedTotal(streamId: bigint | string): Promise<bigint> {
async streamedTotal(streamId: bigint | string, signal?: AbortSignal): Promise<bigint> {
const id = BigInt(streamId);
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
const addr = await this._resolveAddr(id);
const caller = await this._resolveCallerAddress();
const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, caller, addr, 'streamed_total', []);
Expand All @@ -383,8 +386,9 @@
}

/** Withdraw tokens as the recipient. Defaults to full available balance. */
async withdraw(streamId: bigint | string, amount?: bigint): Promise<string> {
async withdraw(streamId: bigint | string, amount?: bigint, signal?: AbortSignal): Promise<string> {
this._ensureCanMutate();
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
const id = BigInt(streamId);
// Fail fast on invalid amounts, the same way create() validates its
// payload client-side, instead of paying for a full simulate+reject
Expand Down Expand Up @@ -592,26 +596,30 @@
}

/** Cancel the stream (sender only). Settles all balances atomically. */
async cancel(streamId: bigint | string): Promise<string> {
async cancel(streamId: bigint | string, signal?: AbortSignal): Promise<string> {
this._ensureCanMutate();
return this._invoke(await this._resolveAddr(BigInt(streamId)), 'cancel', []);
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
return this._invoke(await this._resolveAddr(BigInt(streamId), signal), 'cancel', [], signal);
}

/** Pause the stream (sender only). */
async pause(streamId: bigint | string): Promise<string> {
async pause(streamId: bigint | string, signal?: AbortSignal): Promise<string> {
this._ensureCanMutate();
return this._invoke(await this._resolveAddr(BigInt(streamId)), 'pause', []);
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
return this._invoke(await this._resolveAddr(BigInt(streamId), signal), 'pause', [], signal);
}

/** Resume a paused stream (sender only). Shifts start/end times forward. */
async resume(streamId: bigint | string): Promise<string> {
async resume(streamId: bigint | string, signal?: AbortSignal): Promise<string> {
this._ensureCanMutate();
return this._invoke(await this._resolveAddr(BigInt(streamId)), 'resume', []);
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
return this._invoke(await this._resolveAddr(BigInt(streamId), signal), 'resume', [], signal);
}

/** Deposit additional tokens into the stream (sender only). */
async topUp(streamId: bigint | string, amount: bigint): Promise<string> {
async topUp(streamId: bigint | string, amount: bigint, signal?: AbortSignal): Promise<string> {
this._ensureCanMutate();
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
// Client-side guard mirroring the contract's StreamErrorCode.InvalidAmount
// so a zero/negative top-up fails fast instead of round-tripping.
if (amount <= 0n) {
Expand All @@ -635,8 +643,9 @@
* sender from indefinitely pausing a stream to hold unstreamed tokens
* hostage.
*/
async forceCancel(streamId: bigint | string): Promise<string> {
async forceCancel(streamId: bigint | string, signal?: AbortSignal): Promise<string> {
this._ensureCanMutate();
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
return this._invoke(await this._resolveAddr(BigInt(streamId)), 'force_cancel', []);
}

Expand All @@ -645,8 +654,9 @@
* The new recipient inherits all rights, including the withdrawable
* balance accrued up to the moment of transfer.
*/
async transferRecipient(streamId: bigint | string, newRecipient: string): Promise<string> {
async transferRecipient(streamId: bigint | string, newRecipient: string, signal?: AbortSignal): Promise<string> {
this._ensureCanMutate();
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
if (!newRecipient || typeof newRecipient !== 'string' || !newRecipient.trim()) {
throw new Error('Invalid recipient address: must be a non-empty string');
}
Expand All @@ -659,8 +669,9 @@
* Clawback unstreamed tokens (sender; only if enabled at creation).
* Returns the amount reclaimed (simulated before submission).
*/
async clawback(streamId: bigint | string): Promise<bigint> {
async clawback(streamId: bigint | string, signal?: AbortSignal): Promise<bigint> {
this._ensureCanMutate();
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
const addr = await this._resolveAddr(BigInt(streamId));
const caller = await this._getSenderAddress();
const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, caller, addr, 'clawback', [], this._fee);
Expand All @@ -686,8 +697,9 @@
* Soroban simulation. Returns the resource fee (CPU/RAM), base fee, and
* total estimated fee in stroops.
*/
async estimateFee(operation: StreamOperation): Promise<FeeEstimate> {
async estimateFee(operation: StreamOperation, signal?: AbortSignal): Promise<FeeEstimate> {
const callerAddr = await this._getSenderAddress();
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
const server = this._server();

let tx: Transaction;
Expand Down Expand Up @@ -787,8 +799,9 @@
* Returns a page of StreamInfo along with pagination metadata so the
* frontend can implement infinite scrolling.
*/
async list(params: ListStreamsParams): Promise<PaginatedStreams> {
async list(params: ListStreamsParams, signal?: AbortSignal): Promise<PaginatedStreams> {
const { sender, recipient } = params;
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
// Clamp here (not just in FactoryModule) so hasNextPage/nextCursor math
// below stays consistent with the limit actually sent to the contract —
// otherwise a caller-supplied limit above the max would silently break
Expand Down Expand Up @@ -985,7 +998,8 @@
return this._rpcServerProxy;
}

private async _resolveAddr(id: bigint): Promise<string> {
private async _resolveAddr(id: bigint, signal?: AbortSignal): Promise<string> {
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
// Use the factory's bounded LRU cache for address resolution.
// Stream contract addresses are immutable once assigned by the factory,
// so the cache never needs invalidation.
Expand All @@ -994,7 +1008,8 @@
return addr;
}

private async _simulateTx(tx: Transaction): Promise<xdr.ScVal> {
private async _simulateTx(tx: Transaction, signal?: AbortSignal): Promise<xdr.ScVal> {
if (signal?.aborted) throw new DOMException('Aborted', 'AbortError');
const server = this._server();
const result = await catchNetworkError('simulateTransaction', server.simulateTransaction(tx));
if (SorobanRpc.Api.isSimulationError(result)) {
Expand All @@ -1005,7 +1020,7 @@
}

/** Simulate -> assemble -> sign -> submit -> poll. Returns txHash. */
private async _invoke(contractId: string, method: string, args: xdr.ScVal[]): Promise<string> {
private async _invoke(contractId: string, method: string, args: xdr.ScVal[], signal?: AbortSignal): Promise<string> {
const senderAddr = await this._getSenderAddress();
const tx = await buildContractCallTx(this.rpcUrl, this.passphrase, senderAddr, contractId, method, args, this._fee);
const server = this._server();
Expand All @@ -1015,13 +1030,13 @@
}
const assembled = SorobanRpc.assembleTransaction(tx, sim).build();
const signed = await this._signTx(assembled);
const { hash } = await this._sendAndPoll(server, signed);
const { hash } = await this._sendAndPoll(server, signed, signal);
return hash;
}

private async _sendAndPoll(
server: SorobanRpc.Server,
tx: Transaction,
tx: Transaction, signal?: AbortSignal,
): Promise<{ hash: string; returnValue: xdr.ScVal | undefined }> {
let sent;
try {
Expand All @@ -1037,6 +1052,7 @@
const pollIntervalMs = this.config.confirmationPollIntervalMs ?? DEFAULT_CONFIRMATION_POLL_INTERVAL_MS;
for (let i = 0; i < maxAttempts; i++) {
await sleep(pollIntervalMs);
if (signal?.aborted) throw new DOMException("Aborted", "AbortError");
let s;
try {
s = await catchNetworkError('getTransaction', server.getTransaction(hash));
Expand Down
Loading