Skip to content
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

Jump to bottom

feat(filter)!: return error codes instead of throwing errors #1971

Merged
merged 21 commits into from
May 9, 2024
Merged

feat(filter)!: return error codes instead of throwing errors #1971

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
dd1ce6d
move protocol result type to interfaces
danisharora099 Apr 23, 2024
8226ee3
chore: update type names for verbosity
danisharora099 Apr 23, 2024
2303e79
feat(filter-core): convert error throws to return types
danisharora099 Apr 23, 2024
07ed52f
chore: update types & imports
danisharora099 Apr 23, 2024
f03e86d
update Filter API
danisharora099 Apr 23, 2024
b6c116e
chore: update createSubscription
danisharora099 Apr 23, 2024
2688ec9
chore: update imports & rename
danisharora099 Apr 23, 2024
cf9fda2
chore: update all tests
danisharora099 Apr 23, 2024
05b1baa
chore: resolve conflicts & merge (1/2)
danisharora099 Apr 29, 2024
318bbfd
chore: resolve conflicts & merge (2/n)
danisharora099 Apr 29, 2024
85cb393
chore: resolve conflicts & merge (3/n)
danisharora099 Apr 29, 2024
ffb9b2d
chore: resolve conflicts & merge (4/n)
danisharora099 Apr 29, 2024
891ed66
chore: resolve conflicts & merge (5/n)
danisharora099 Apr 29, 2024
fc1cacf
chore: resolve conflicts & merge (6/n)
danisharora099 Apr 29, 2024
4e37405
Merge branch 'master' of github.com:waku-org/js-waku into feat/filter…
danisharora099 Apr 30, 2024
f2d6d8f
chore: use idiomatic approach
danisharora099 May 7, 2024
f48c58d
chore: fix tests
danisharora099 May 7, 2024
fd7d61c
chore: address comments
danisharora099 May 7, 2024
0f7d3c2
chore: fix test
danisharora099 May 7, 2024
1f6e420
Merge branch 'master' into feat/filter-error-codes
danisharora099 May 9, 2024
365fae0
rm: only
danisharora099 May 9, 2024
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
206 changes: 157 additions & 49 deletions packages/core/src/lib/filter/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,17 +1,20 @@
import type { Peer } from "@libp2p/interface";
import type { Peer, Stream } from "@libp2p/interface";
import type { IncomingStreamData } from "@libp2p/interface-internal";
import type {
ContentTopic,
IBaseProtocolCore,
Libp2p,
ProtocolCreateOptions,
PubsubTopic
import {
type ContentTopic,
type CoreProtocolResult,
type IBaseProtocolCore,
type Libp2p,
type ProtocolCreateOptions,
ProtocolError,
type PubsubTopic
} from "@waku/interfaces";
import { WakuMessage } from "@waku/proto";
import { Logger } from "@waku/utils";
import all from "it-all";
import * as lp from "it-length-prefixed";
import { pipe } from "it-pipe";
import { Uint8ArrayList } from "uint8arraylist";

import { BaseProtocol } from "../base_protocol.js";

Expand Down Expand Up @@ -90,53 +93,106 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
pubsubTopic: PubsubTopic,
peer: Peer,
contentTopics: ContentTopic[]
): Promise<void> {
): Promise<CoreProtocolResult> {
const stream = await this.getStream(peer);

const request = FilterSubscribeRpc.createSubscribeRequest(
pubsubTopic,
contentTopics
);

const res = await pipe(
[request.encode()],
lp.encode,
stream,
lp.decode,
async (source) => await all(source)
);

if (!res || !res.length) {
throw Error(
`No response received for request ${request.requestId}: ${res}`
let res: Uint8ArrayList[] | undefined;
try {
res = await pipe(
[request.encode()],
lp.encode,
stream,
lp.decode,
async (source) => await all(source)
);
} catch (error) {
log.error("Failed to send subscribe request", error);
return {
success: null,
failure: {
error: ProtocolError.GENERIC_FAIL,
peerId: peer.id
}
};
}

const { statusCode, requestId, statusDesc } =
FilterSubscribeResponse.decode(res[0].slice());

if (statusCode < 200 || statusCode >= 300) {
throw new Error(
log.error(
`Filter subscribe request ${requestId} failed with status code ${statusCode}: ${statusDesc}`
);
return {
failure: {
error: ProtocolError.REMOTE_PEER_REJECTED,
peerId: peer.id
},
success: null
};
}

return {
failure: null,
success: peer.id
};
}

async unsubscribe(
pubsubTopic: PubsubTopic,
peer: Peer,
contentTopics: ContentTopic[]
): Promise<void> {
const stream = await this.getStream(peer);
): Promise<CoreProtocolResult> {
let stream: Stream | undefined;
try {
stream = await this.getStream(peer);
} catch (error) {
log.error(
`Failed to get a stream for remote peer${peer.id.toString()}`,
error
);
return {
success: null,
failure: {
error: ProtocolError.REMOTE_PEER_FAULT,
peerId: peer.id
}
};
}

const unsubscribeRequest = FilterSubscribeRpc.createUnsubscribeRequest(
pubsubTopic,
contentTopics
);

await pipe([unsubscribeRequest.encode()], lp.encode, stream.sink);
try {
await pipe([unsubscribeRequest.encode()], lp.encode, stream.sink);
} catch (error) {
log.error("Failed to send unsubscribe request", error);
return {
success: null,
failure: {
error: ProtocolError.GENERIC_FAIL,
peerId: peer.id
Copy link
Collaborator

Choose a reason for hiding this comment

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

shouldn't it be more concrete that generic?

Copy link
Collaborator

Choose a reason for hiding this comment

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

I think we should make errors as informative as possible (without creating a mess) as it is the only thing developer would get, so let's try not to use GENERIC_FAIL imho

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

shouldn't it be more concrete that generic?

no, this is also what we do during any lightpush request.

if pipe fails unexpectedly, we can't be sure of what went wrong.

Copy link
Collaborator

Choose a reason for hiding this comment

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

is there no response that we can read? I expect full node potentially send some message if it is on the side of it, but if failure is in pipe - then it can be generic (though in that case I still think there is a reason we can root cause)

can you repro unexpected fails of pipe?

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

can you repro unexpected fails of pipe?

no test vector as such for that

--

this try catch case actually handles error if error is thrown and there is no defined response from the pipe operation

if you follow the next few lines you will see we also handle errors based on bad values of res

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

eg:

    try {
      res = await pipe(
        [query.encode()],
        lp.encode,
        stream,
        lp.decode,
        async (source) => await all(source)
      );
    } catch (err) {
      log.error("Failed to send waku light push request", err);
      return {
        success: null,
        failure: {
          error: ProtocolError.GENERIC_FAIL,
          peerId: peer.id
        }
      };
    }

    const bytes = new Uint8ArrayList();
    res.forEach((chunk) => {
      bytes.append(chunk);
    });

    let response: PushResponse | undefined;
    try {
      response = PushRpc.decode(bytes).response;
    } catch (err) {
      log.error("Failed to decode push reply", err);
      return {
        success: null,
        failure: {
          error: ProtocolError.DECODE_FAILED,
          peerId: peer.id
        }
      };
    }

    if (!response) {
      log.error("Remote peer fault: No response in PushRPC");
      return {
        success: null,
        failure: {
          error: ProtocolError.REMOTE_PEER_FAULT,
          peerId: peer.id
        }
      };
    }

    if (!response.isSuccess) {
      log.error("Remote peer rejected the message: ", response.info);
      return {
        success: null,
        failure: {
          error: ProtocolError.REMOTE_PEER_REJECTED,
          peerId: peer.id
        }
      };
    }

Copy link
Collaborator

Choose a reason for hiding this comment

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

I believe in that case GENERIC_FAIL on pipe means - failed to send a request, as it fails locally.

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

I believe in that case GENERIC_FAIL on pipe means - failed to send a request, as it fails locally.

hmm, could it fail while receiving a response for the remote peer?
are we sure about that?

Copy link
Collaborator

Choose a reason for hiding this comment

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

dunno, I think we need to investigate it - if it bothers us

}
};
}

return {
success: peer.id,
failure: null
};
}

async unsubscribeAll(pubsubTopic: PubsubTopic, peer: Peer): Promise<void> {
async unsubscribeAll(
pubsubTopic: PubsubTopic,
peer: Peer
): Promise<CoreProtocolResult> {
const stream = await this.getStream(peer);

const request = FilterSubscribeRpc.createUnsubscribeAllRequest(pubsubTopic);
Expand All @@ -150,53 +206,105 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
);

if (!res || !res.length) {
throw Error(
`No response received for request ${request.requestId}: ${res}`
);
return {
failure: {
error: ProtocolError.REMOTE_PEER_FAULT,
peerId: peer.id
},
success: null
};
}

const { statusCode, requestId, statusDesc } =
FilterSubscribeResponse.decode(res[0].slice());

if (statusCode < 200 || statusCode >= 300) {
throw new Error(
log.error(
`Filter unsubscribe all request ${requestId} failed with status code ${statusCode}: ${statusDesc}`
);
return {
failure: {
error: ProtocolError.REMOTE_PEER_REJECTED,
peerId: peer.id
Copy link
Collaborator

Choose a reason for hiding this comment

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

here and in similar places - can we map statusCode to some tangible error

like in case if full node has no capacity (ddos etc) - it might return some particular code

to cover this I think we should do follow up, and sync with @NagyZoltanPeter of what error codes are expected here and in similar places

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

agree - that would be useful.

we do follow generic error handling based on statusCode:

 const { statusCode, requestId, statusDesc } =
      FilterSubscribeResponse.decode(res[0].slice());

    if (statusCode < 200 || statusCode >= 300) {
      log.error(
        `Filter unsubscribe all request ${requestId} failed with status code ${statusCode}: ${statusDesc}`
      );
      return {
        failure: {
          error: ProtocolError.REMOTE_PEER_REJECTED,
          peerId: peer.id
        },
        success: null
      };
    }

creating an issue to track this: #1999

},
success: null
};
}

return {
failure: null,
success: peer.id
};
}

async ping(peer: Peer): Promise<void> {
const stream = await this.getStream(peer);
async ping(peer: Peer): Promise<CoreProtocolResult> {
let stream: Stream | undefined;
try {
stream = await this.getStream(peer);
} catch (error) {
log.error(
`Failed to get a stream for remote peer${peer.id.toString()}`,
error
);
return {
success: null,
failure: {
error: ProtocolError.REMOTE_PEER_FAULT,
peerId: peer.id
}
};
}

const request = FilterSubscribeRpc.createSubscriberPingRequest();

let res: Uint8ArrayList[] | undefined;
try {
const res = await pipe(
res = await pipe(
[request.encode()],
lp.encode,
stream,
lp.decode,
async (source) => await all(source)
);

if (!res || !res.length) {
throw Error(
`No response received for request ${request.requestId}: ${res}`
);
}

const { statusCode, requestId, statusDesc } =
FilterSubscribeResponse.decode(res[0].slice());

if (statusCode < 200 || statusCode >= 300) {
throw new Error(
`Filter ping request ${requestId} failed with status code ${statusCode}: ${statusDesc}`
);
}
log.info(`Ping successful for peer ${peer.id.toString()}`);
} catch (error) {
log.error("Error pinging: ", error);
throw error; // Rethrow the actual error instead of wrapping it
log.error("Failed to send ping request", error);
return {
success: null,
failure: {
error: ProtocolError.GENERIC_FAIL,
peerId: peer.id
}
};
}

if (!res || !res.length) {
return {
success: null,
failure: {
error: ProtocolError.REMOTE_PEER_FAULT,
peerId: peer.id
}
};
}

const { statusCode, requestId, statusDesc } =
FilterSubscribeResponse.decode(res[0].slice());

if (statusCode < 200 || statusCode >= 300) {
log.error(
`Filter ping request ${requestId} failed with status code ${statusCode}: ${statusDesc}`
);
return {
success: null,
failure: {
error: ProtocolError.REMOTE_PEER_REJECTED,
peerId: peer.id
}
};
}
return {
success: peer.id,
failure: null
};
}
}
22 changes: 10 additions & 12 deletions packages/core/src/lib/light_push/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,13 @@
import type { Peer, PeerId, Stream } from "@libp2p/interface";
import type { Peer, Stream } from "@libp2p/interface";
import {
Failure,
IBaseProtocolCore,
IEncoder,
IMessage,
Libp2p,
ProtocolCreateOptions,
type CoreProtocolResult,
type IBaseProtocolCore,
type IEncoder,
type IMessage,
type Libp2p,
type ProtocolCreateOptions,
ProtocolError,
ProtocolResult
type ThisOrThat
} from "@waku/interfaces";
import { PushResponse } from "@waku/proto";
import { isMessageSizeUnderCap } from "@waku/utils";
Expand All @@ -26,9 +26,7 @@ const log = new Logger("light-push");
export const LightPushCodec = "/vac/waku/lightpush/2.0.0-beta1";
export { PushResponse };

type PreparePushMessageResult = ProtocolResult<"query", PushRpc>;

type CoreSendResult = ProtocolResult<"success", PeerId, "failure", Failure>;
type PreparePushMessageResult = ThisOrThat<"query", PushRpc>;

/**
* Implements the [Waku v2 Light Push protocol](https://rfc.vac.dev/spec/19/).
Expand Down Expand Up @@ -84,7 +82,7 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
encoder: IEncoder,
message: IMessage,
peer: Peer
): Promise<CoreSendResult> {
): Promise<CoreProtocolResult> {
const { query, error: preparationError } = await this.preparePushMessage(
encoder,
message
Expand Down
10 changes: 6 additions & 4 deletions packages/core/src/lib/metadata/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,9 @@ import { IncomingStreamData } from "@libp2p/interface";
import {
type IMetadata,
type Libp2pComponents,
type MetadataQueryResult,
type PeerIdStr,
ProtocolError,
QueryResult,
type ShardInfo
} from "@waku/interfaces";
import { proto_metadata } from "@waku/proto";
Expand Down Expand Up @@ -74,7 +74,7 @@ class Metadata extends BaseProtocol implements IMetadata {
/**
* Make a metadata query to a peer
*/
async query(peerId: PeerId): Promise<QueryResult> {
async query(peerId: PeerId): Promise<MetadataQueryResult> {
const request = proto_metadata.WakuMetadataRequest.encode(this.shardInfo);

const peer = await this.peerStore.get(peerId);
Expand Down Expand Up @@ -112,7 +112,9 @@ class Metadata extends BaseProtocol implements IMetadata {
};
}

public async confirmOrAttemptHandshake(peerId: PeerId): Promise<QueryResult> {
public async confirmOrAttemptHandshake(
peerId: PeerId
): Promise<MetadataQueryResult> {
const shardInfo = this.handshakesConfirmed.get(peerId.toString());
if (shardInfo) {
return {
Expand All @@ -126,7 +128,7 @@ class Metadata extends BaseProtocol implements IMetadata {

private decodeMetadataResponse(
encodedResponse: Uint8ArrayList[]
): QueryResult {
): MetadataQueryResult {
const bytes = new Uint8ArrayList();

encodedResponse.forEach((chunk) => {
Expand Down
6 changes: 4 additions & 2 deletions packages/discovery/src/peer-exchange/waku_peer_exchange.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import {
IPeerExchange,
Libp2pComponents,
PeerExchangeQueryParams,
PeerExchangeResult,
PeerExchangeQueryResult,
ProtocolError,
PubsubTopic
} from "@waku/interfaces";
Expand Down Expand Up @@ -35,7 +35,9 @@ export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
/**
* Make a peer exchange query to a peer
*/
async query(params: PeerExchangeQueryParams): Promise<PeerExchangeResult> {
async query(
params: PeerExchangeQueryParams
): Promise<PeerExchangeQueryResult> {
const { numPeers } = params;
const rpcQuery = PeerExchangeRPC.createRequest({
numPeers: BigInt(numPeers)
Expand Down
Loading
Loading