Skip to content

Chore/abort controller #629

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

Closed
ivan-angjelkoski wants to merge 3 commits into from
Closed

Chore/abort controller #629

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3936fb9
feat: abort controller grpc
ivan-angjelkoski Mar 24, 2026
413c5d1
chore: abort controller http rest
ivan-angjelkoski Mar 24, 2026
982655b
chore: package lock
ivan-angjelkoski Mar 24, 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
33 changes: 27 additions & 6 deletions packages/sdk-ts/src/client/abacus/grpc/AbacusGrpcApi.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import * as PointsSvcPb from '@injectivelabs/abacus-proto-ts-v2/generated/points
import { PointsSvcClient } from '@injectivelabs/abacus-proto-ts-v2/generated/points_svc_pb.client'
import BaseGrpcConsumer from '../../base/BaseGrpcConsumer.js'
import { AbacusGrpcTransformer } from './transformers/index.js'
import type { GrpcCallOptions } from '../../../types/index.js'

export class AbacusGrpcApi extends BaseGrpcConsumer {
protected module: string = IndexerErrorModule.Abacus
Expand All @@ -11,20 +12,28 @@ export class AbacusGrpcApi extends BaseGrpcConsumer {
return this.initClient(PointsSvcClient)
}

async fetchAccountLatestPoints(address: string) {
async fetchAccountLatestPoints(address: string, options?: GrpcCallOptions) {
const request = PointsSvcPb.PointsLatestForAccountRequest.create({
accountAddress: address,
})

const response = await this.executeGrpcCall<
PointsSvcPb.PointsLatestForAccountRequest,
PointsSvcPb.PointsLatestForAccountResponse
>(request, this.client.pointsLatestForAccount.bind(this.client))
>(
request,
this.client.pointsLatestForAccount.bind(this.client),
options?.signal,
)

return AbacusGrpcTransformer.grpcPointsLatestToPointsLatest(response)
}

async fetchAccountDailyPoints(address: string, daysLimit?: number) {
async fetchAccountDailyPoints(
address: string,
daysLimit?: number,
options?: GrpcCallOptions,
) {
const request = PointsSvcPb.PointsStatsDailyForAccountRequest.create({
accountAddress: address,
daysLimit: daysLimit ? BigInt(daysLimit) : undefined,
Expand All @@ -33,14 +42,22 @@ export class AbacusGrpcApi extends BaseGrpcConsumer {
const response = await this.executeGrpcCall<
PointsSvcPb.PointsStatsDailyForAccountRequest,
PointsSvcPb.HistoricalPointsStatsRowCollection
>(request, this.client.pointsStatsDailyForAccount.bind(this.client))
>(
request,
this.client.pointsStatsDailyForAccount.bind(this.client),
options?.signal,
)

return AbacusGrpcTransformer.grpcPointsStatsDailyToPointsStatsDaily(
response,
)
}

async fetchAccountWeeklyPoints(address: string, weeksLimit?: number) {
async fetchAccountWeeklyPoints(
address: string,
weeksLimit?: number,
options?: GrpcCallOptions,
) {
const request = PointsSvcPb.PointsStatsWeeklyForAccountRequest.create({
accountAddress: address,
weeksLimit: weeksLimit ? BigInt(weeksLimit) : undefined,
Expand All @@ -49,7 +66,11 @@ export class AbacusGrpcApi extends BaseGrpcConsumer {
const response = await this.executeGrpcCall<
PointsSvcPb.PointsStatsWeeklyForAccountRequest,
PointsSvcPb.HistoricalPointsStatsRowCollection
>(request, this.client.pointsStatsWeeklyForAccount.bind(this.client))
>(
request,
this.client.pointsStatsWeeklyForAccount.bind(this.client),
options?.signal,
)

return AbacusGrpcTransformer.grpcPointsStatsWeeklyToPointsStatsWeekly(
response,
Expand Down
71 changes: 61 additions & 10 deletions packages/sdk-ts/src/client/base/BaseGrpcConsumer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { RpcError } from '@protobuf-ts/runtime-rpc'
import {
GrpcErrorCode,
UnspecifiedErrorCode,
TransactionException,
grpcErrorCodeToErrorCode,
Expand Down Expand Up @@ -79,14 +80,18 @@ export default class BaseGrpcConsumer {
}

/**
* Builds RpcOptions with metadata
* Builds RpcOptions with metadata and optional abort signal
* @deprecated Options should be managed externally and passed into the constructor instead
*/
protected getRpcOptions(): RpcOptions {
protected getRpcOptions(signal?: AbortSignal): RpcOptions {
const options: RpcOptions = {
meta: this.metadata || {},
}

if (signal) {
options.abort = signal
}

return options
}

Expand All @@ -97,21 +102,53 @@ export default class BaseGrpcConsumer {
grpcCall: () => Promise<TResponse>,
retries: number = 3,
delay: number = 1000,
signal?: AbortSignal,
): Promise<TResponse> {
const retryGrpcCall = async (attempt = 1): Promise<TResponse> => {
if (signal?.aborted) {
throw (
signal.reason ??
new DOMException('The operation was aborted.', 'AbortError')
)
}

try {
return await grpcCall()
} catch (e: unknown) {
if (signal?.aborted) {
throw (
signal.reason ??
new DOMException('The operation was aborted.', 'AbortError')
)
}

if (attempt >= retries) {
throw e
}

return new Promise((resolve) =>
setTimeout(
return new Promise<TResponse>((resolve, reject) => {
const timeoutId = setTimeout(
() => resolve(retryGrpcCall(attempt + 1)),
delay * attempt,
),
)
)

if (signal) {
signal.addEventListener(
'abort',
() => {
clearTimeout(timeoutId)
reject(
signal.reason ??
new DOMException(
'The operation was aborted.',
'AbortError',
),
)
},
{ once: true },
)
}
})
Comment on lines +129 to +151

@coderabbitai coderabbitai Bot Mar 24, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Potential memory leak: abort listener is not removed on successful retry or timeout.

When the retry succeeds or the timeout fires naturally, the abort listener registered on the signal is not cleaned up. If the same AbortController is reused across multiple calls, listeners accumulate.

🛡️ Proposed fix to clean up the abort listener 
         return new Promise<TResponse>((resolve, reject) => {
+          const onAbort = () => {
+            clearTimeout(timeoutId)
+            reject(
+              signal.reason ??
+                new DOMException(
+                  'The operation was aborted.',
+                  'AbortError',
+                ),
+            )
+          }
+
           const timeoutId = setTimeout(
-            () => resolve(retryGrpcCall(attempt + 1)),
+            () => {
+              signal?.removeEventListener('abort', onAbort)
+              resolve(retryGrpcCall(attempt + 1))
+            },
             delay * attempt,
           )

           if (signal) {
-            signal.addEventListener(
-              'abort',
-              () => {
-                clearTimeout(timeoutId)
-                reject(
-                  signal.reason ??
-                    new DOMException(
-                      'The operation was aborted.',
-                      'AbortError',
-                    ),
-                )
-              },
-              { once: true },
-            )
+            signal.addEventListener('abort', onAbort, { once: true })
           }
         })
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@packages/sdk-ts/src/client/base/BaseGrpcConsumer.ts` around lines 129 - 151,
The abort listener added to the AbortSignal in BaseGrpcConsumer.retryGrpcCall is
not removed when the timeout fires or when the retry resolves, which can leak
listeners if the same AbortController is reused; fix by using a named handler
(e.g., const onAbort = () => { ... }) passed to signal.addEventListener and call
signal.removeEventListener(onAbort) before resolving the Promise in the timeout
callback (clearTimeout already exists) and also remove the listener whenever you
resolve or reject the Promise from other code paths so the onAbort handler is
always detached.

}
}

Expand Down Expand Up @@ -155,6 +192,14 @@ export default class BaseGrpcConsumer {
* Otherwise throws a GrpcUnaryRequestException for generic gRPC errors.
*/
protected handleGrpcError(e: unknown, context: string): never {
if (e instanceof DOMException && e.name === 'AbortError') {
throw new GrpcUnaryRequestException(new Error(e.message), {
code: GrpcErrorCode.Canceled,
context,
contextModule: this.module,
})
}

if (e instanceof RpcError) {
const message = e.message
const abciCode = this.getABCICodeFromMessage(message)
Expand Down Expand Up @@ -223,12 +268,18 @@ export default class BaseGrpcConsumer {
req: TRequest,
options?: RpcOptions,
) => UnaryCall<TRequest, TResponse>,
signal?: AbortSignal,
): Promise<TResponse> {
try {
return await this.retry(async () => {
const call = clientMethod(request, this.getRpcOptions())
return await call.response
})
return await this.retry(
async () => {
const call = clientMethod(request, this.getRpcOptions(signal))
return await call.response
},
3,
1000,
signal,
)
} catch (e: unknown) {
// Derive context from method name if not provided
const errorContext = clientMethod.name || 'UnknownMethod'
Expand Down
25 changes: 17 additions & 8 deletions packages/sdk-ts/src/client/chain/grpc/ChainGrpcAuctionApi.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import { QueryClient as InjectiveAuctionV1Beta1QueryClient } from '@injectivelab
import { ChainModule } from '../types/index.js'
import BaseGrpcConsumer from '../../base/BaseGrpcConsumer.js'
import { ChainGrpcAuctionTransformer } from '../transformers/index.js'
import type { GrpcCallOptions } from '../../../types/index.js'
/**
* @category Chain Grpc API
*/
Expand All @@ -13,56 +14,64 @@ export class ChainGrpcAuctionApi extends BaseGrpcConsumer {
return this.initClient(InjectiveAuctionV1Beta1QueryClient)
}

async fetchModuleParams() {
async fetchModuleParams(options?: GrpcCallOptions) {
const request =
InjectiveAuctionV1Beta1QueryPb.QueryAuctionParamsRequest.create()

const response = await this.executeGrpcCall<
InjectiveAuctionV1Beta1QueryPb.QueryAuctionParamsRequest,
InjectiveAuctionV1Beta1QueryPb.QueryAuctionParamsResponse
>(request, this.client.auctionParams.bind(this.client))
>(request, this.client.auctionParams.bind(this.client), options?.signal)

return ChainGrpcAuctionTransformer.moduleParamsResponseToModuleParams(
response,
)
}

async fetchCurrentBasket() {
async fetchCurrentBasket(options?: GrpcCallOptions) {
const request =
InjectiveAuctionV1Beta1QueryPb.QueryCurrentAuctionBasketRequest.create()

const response = await this.executeGrpcCall<
InjectiveAuctionV1Beta1QueryPb.QueryCurrentAuctionBasketRequest,
InjectiveAuctionV1Beta1QueryPb.QueryCurrentAuctionBasketResponse
>(request, this.client.currentAuctionBasket.bind(this.client))
>(
request,
this.client.currentAuctionBasket.bind(this.client),
options?.signal,
)

return ChainGrpcAuctionTransformer.currentBasketResponseToCurrentBasket(
response,
)
}

async fetchModuleState() {
async fetchModuleState(options?: GrpcCallOptions) {
const request =
InjectiveAuctionV1Beta1QueryPb.QueryModuleStateRequest.create()

const response = await this.executeGrpcCall<
InjectiveAuctionV1Beta1QueryPb.QueryModuleStateRequest,
InjectiveAuctionV1Beta1QueryPb.QueryModuleStateResponse
>(request, this.client.auctionModuleState.bind(this.client))
>(
request,
this.client.auctionModuleState.bind(this.client),
options?.signal,
)

return ChainGrpcAuctionTransformer.auctionModuleStateResponseToAuctionModuleState(
response,
)
}

async fetchLastAuctionResult() {
async fetchLastAuctionResult(options?: GrpcCallOptions) {
const request =
InjectiveAuctionV1Beta1QueryPb.QueryLastAuctionResultRequest.create()

const response = await this.executeGrpcCall<
InjectiveAuctionV1Beta1QueryPb.QueryLastAuctionResultRequest,
InjectiveAuctionV1Beta1QueryPb.QueryLastAuctionResultResponse
>(request, this.client.lastAuctionResult.bind(this.client))
>(request, this.client.lastAuctionResult.bind(this.client), options?.signal)

return ChainGrpcAuctionTransformer.lastAuctionResultResponseToLastAuctionResult(
response,
Expand Down
16 changes: 10 additions & 6 deletions packages/sdk-ts/src/client/chain/grpc/ChainGrpcAuthApi.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import { ChainModule } from '../types/index.js'
import BaseGrpcConsumer from '../../base/BaseGrpcConsumer.js'
import { ChainGrpcAuthTransformer } from '../transformers/ChainGrpcAuthTransformer.js'
import { ChainGrpcCommonTransformer } from '../transformers/ChainGrpcCommonTransformer.js'
import type { GrpcCallOptions } from '../../../types/index.js'
import type { PaginationOption } from '../../../types/pagination.js'
/**
* @category Chain Grpc API
Expand All @@ -15,31 +16,34 @@ export class ChainGrpcAuthApi extends BaseGrpcConsumer {
return this.initClient(CosmosAuthV1BetaQueryClient)
}

async fetchModuleParams() {
async fetchModuleParams(options?: GrpcCallOptions) {
const request = CosmosAuthV1Beta1QueryPb.QueryParamsRequest.create()

const response = await this.executeGrpcCall<
CosmosAuthV1Beta1QueryPb.QueryParamsRequest,
CosmosAuthV1Beta1QueryPb.QueryParamsResponse
>(request, this.client.params.bind(this.client))
>(request, this.client.params.bind(this.client), options?.signal)

return ChainGrpcAuthTransformer.moduleParamsResponseToModuleParams(response)
}

async fetchAccount(address: string) {
async fetchAccount(address: string, options?: GrpcCallOptions) {
const request = CosmosAuthV1Beta1QueryPb.QueryAccountRequest.create()

request.address = address

const response = await this.executeGrpcCall<
CosmosAuthV1Beta1QueryPb.QueryAccountRequest,
CosmosAuthV1Beta1QueryPb.QueryAccountResponse
>(request, this.client.account.bind(this.client))
>(request, this.client.account.bind(this.client), options?.signal)

return ChainGrpcAuthTransformer.accountResponseToAccount(response)
}

async fetchAccounts(pagination?: PaginationOption) {
async fetchAccounts(
pagination?: PaginationOption,
options?: GrpcCallOptions,
) {
const request = CosmosAuthV1Beta1QueryPb.QueryAccountsRequest.create()
const paginationForRequest =
ChainGrpcCommonTransformer.pageRequestToGrpcPageRequestV2(pagination)
Expand All @@ -51,7 +55,7 @@ export class ChainGrpcAuthApi extends BaseGrpcConsumer {
const response = await this.executeGrpcCall<
CosmosAuthV1Beta1QueryPb.QueryAccountsRequest,
CosmosAuthV1Beta1QueryPb.QueryAccountsResponse
>(request, this.client.accounts.bind(this.client))
>(request, this.client.accounts.bind(this.client), options?.signal)

return ChainGrpcAuthTransformer.accountsResponseToAccounts(response)
}
Expand Down
Loading