Compare commits
11 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Renato Alcara
|
dc24e60e5b | ||
|
Renato Alcara
|
942f4ae2c9 | ||
|
Renato Alcara
|
98d2de2984 | ||
|
Renato Alcara
|
b0c34cb536 | ||
|
Renato Alcara
|
d169a6f755 | ||
|
Renato Alcara
|
aef4973cac | ||
|
Renato Alcara
|
b90b383811 | ||
|
Renato Alcara
|
35461f96c5 | ||
|
Renato Alcara
|
1e37448e05 | ||
|
Renato Alcara
|
72e78a12e0 | ||
|
Renato Alcara
|
a5b7ce7ef9 |
+1
-163
@@ -1,7 +1,7 @@
|
||||
syntax = "proto3";
|
||||
package proto;
|
||||
|
||||
/// WhatsApp Version: 2.3000.1038569661
|
||||
/// WhatsApp Version: 2.3000.1038164556
|
||||
|
||||
message ADVDeviceIdentity {
|
||||
optional uint32 rawId = 1;
|
||||
@@ -14,7 +14,6 @@ message ADVDeviceIdentity {
|
||||
enum ADVEncryptionType {
|
||||
E2EE = 0;
|
||||
HOSTED = 1;
|
||||
NON_E2EE = 2;
|
||||
}
|
||||
message ADVKeyIndexList {
|
||||
optional uint32 rawId = 1;
|
||||
@@ -62,7 +61,6 @@ message AIHomeState {
|
||||
ANIMATE_PHOTO = 2;
|
||||
ANALYZE_FILE = 3;
|
||||
COLLABORATE = 4;
|
||||
OPEN_GREETING_CARD = 5;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -249,16 +247,6 @@ message AIRichResponseUnifiedResponse {
|
||||
optional bytes data = 1;
|
||||
}
|
||||
|
||||
enum AISubscriptionRequestType {
|
||||
UNSPECIFIED = 0;
|
||||
THINK_HARD = 1;
|
||||
IMAGE_GEN = 2;
|
||||
VIDEO_GEN = 3;
|
||||
}
|
||||
message AISubscriptionUpsellMetadata {
|
||||
optional AISubscriptionRequestType requestType = 1;
|
||||
}
|
||||
|
||||
message AIThreadInfo {
|
||||
optional AIThreadServerInfo serverInfo = 1;
|
||||
optional AIThreadClientInfo clientInfo = 2;
|
||||
@@ -671,8 +659,6 @@ message BotMetadata {
|
||||
optional BotInfrastructureDiagnostics botInfrastructureDiagnostics = 37;
|
||||
optional AIMediaCollectionMetadata aiMediaCollectionMetadata = 38;
|
||||
optional BotCommandMetadata commandMetadata = 39;
|
||||
optional BotResolvedToolCallMetadata resolvedToolCallMetadata = 40;
|
||||
optional AISubscriptionUpsellMetadata subscriptionUpsellMetadata = 41;
|
||||
optional bytes internalMetadata = 999;
|
||||
}
|
||||
|
||||
@@ -911,11 +897,6 @@ message BotRenderingMetadata {
|
||||
|
||||
}
|
||||
|
||||
message BotResolvedToolCallMetadata {
|
||||
optional string toolCallId = 1;
|
||||
optional string resolutionDataSerialized = 2;
|
||||
}
|
||||
|
||||
message BotSessionMetadata {
|
||||
optional string sessionId = 1;
|
||||
optional BotSessionSource sessionSource = 2;
|
||||
@@ -1151,7 +1132,6 @@ message ClientPayload {
|
||||
optional int32 preacksCount = 45;
|
||||
optional int32 processingQueueSize = 46;
|
||||
repeated string pairedPeripherals = 47;
|
||||
optional bytes testIsolationId = 48;
|
||||
enum AccountType {
|
||||
DEFAULT = 0;
|
||||
GUEST = 1;
|
||||
@@ -1247,7 +1227,6 @@ message ClientPayload {
|
||||
optional string deviceExpId = 14;
|
||||
optional DeviceType deviceType = 15;
|
||||
optional string deviceModelType = 16;
|
||||
optional DistributionChannel distributionChannel = 17;
|
||||
message AppVersion {
|
||||
optional uint32 primary = 1;
|
||||
optional uint32 secondary = 2;
|
||||
@@ -1263,12 +1242,6 @@ message ClientPayload {
|
||||
WEARABLE = 3;
|
||||
VR = 4;
|
||||
}
|
||||
enum DistributionChannel {
|
||||
APPSTORE = 0;
|
||||
WEBSITE = 1;
|
||||
TESTFLIGHT = 2;
|
||||
INTERNAL = 3;
|
||||
}
|
||||
enum Platform {
|
||||
ANDROID = 0;
|
||||
IOS = 1;
|
||||
@@ -1438,9 +1411,6 @@ message ContextInfo {
|
||||
optional MediaDomainInfo mediaDomainInfo = 74;
|
||||
optional PartiallySelectedContent partiallySelectedContent = 75;
|
||||
optional uint32 afterReadDuration = 76;
|
||||
optional CrossAppSource crossAppSource = 77;
|
||||
optional BusinessInteractionPills businessInteractionPills = 78;
|
||||
optional string posterStatusId = 79;
|
||||
message AdReplyInfo {
|
||||
optional string advertiserName = 1;
|
||||
optional MediaType mediaType = 2;
|
||||
@@ -1453,47 +1423,10 @@ message ContextInfo {
|
||||
}
|
||||
}
|
||||
|
||||
message BusinessInteractionPills {
|
||||
optional string businessJid = 1;
|
||||
repeated Pill pills = 2;
|
||||
optional EntryPoint entryPoint = 3;
|
||||
enum EntryPoint {
|
||||
ENTRY_POINT_UNKNOWN = 0;
|
||||
P2P_LINK_SHARE = 1;
|
||||
CONTACT_CARD_SHARING = 2;
|
||||
PHONE_NUMBER = 3;
|
||||
STATUS = 4;
|
||||
IN_THREAD_CONTEXT_CARD = 5;
|
||||
}
|
||||
message Pill {
|
||||
optional ContextInfo.BusinessInteractionPills.PillType pillType = 1;
|
||||
optional string actionUrl = 2;
|
||||
}
|
||||
|
||||
enum PillType {
|
||||
UNKNOWN = 0;
|
||||
VIEW_BUSINESS = 1;
|
||||
CHAT = 2;
|
||||
CALL = 3;
|
||||
CATALOG = 4;
|
||||
CHANNEL = 5;
|
||||
BOOK_APPOINTMENT = 6;
|
||||
OFFERS = 7;
|
||||
BESTSELLERS = 8;
|
||||
MENU = 9;
|
||||
ABOUT = 10;
|
||||
}
|
||||
}
|
||||
|
||||
message BusinessMessageForwardInfo {
|
||||
optional string businessOwnerJid = 1;
|
||||
}
|
||||
|
||||
enum CrossAppSource {
|
||||
CROSS_APP_SOURCE_UNKNOWN = 0;
|
||||
CROSS_APP_SOURCE_INSTAGRAM = 1;
|
||||
CROSS_APP_SOURCE_FACEBOOK = 2;
|
||||
}
|
||||
message DataSharingContext {
|
||||
optional bool showMmDisclosure = 1;
|
||||
optional string encryptedSignalTokenConsented = 2;
|
||||
@@ -1702,7 +1635,6 @@ message Conversation {
|
||||
optional bool isMarketingMessageThread = 55;
|
||||
optional bool isSenderNewAccount = 56;
|
||||
optional uint32 afterReadDuration = 57;
|
||||
optional bool isSenderSuspicious = 58;
|
||||
enum EndOfHistoryTransferType {
|
||||
COMPLETE_BUT_MORE_MESSAGES_REMAIN_ON_PRIMARY = 0;
|
||||
COMPLETE_AND_NO_MORE_MESSAGE_REMAIN_ON_PRIMARY = 1;
|
||||
@@ -2416,8 +2348,6 @@ message Message {
|
||||
optional PollAddOptionMessage pollAddOptionMessage = 121;
|
||||
optional EventInviteMessage eventInviteMessage = 122;
|
||||
optional GroupRootKeyShare groupRootKeyShare = 123;
|
||||
optional P2PPaymentReminderNotification p2PPaymentReminderNotification = 124;
|
||||
optional SplitPaymentMessage splitPaymentMessage = 125;
|
||||
message AlbumMessage {
|
||||
optional uint32 expectedImageCount = 2;
|
||||
optional uint32 expectedVideoCount = 3;
|
||||
@@ -2598,41 +2528,6 @@ message Message {
|
||||
optional string id = 2;
|
||||
}
|
||||
|
||||
message ChatCustomImageWallpaper {
|
||||
optional string directPath = 1;
|
||||
optional bytes mediaKey = 2;
|
||||
optional bytes fileEncSha256 = 3;
|
||||
optional bytes fileSha256 = 4;
|
||||
optional float dimLevel = 5;
|
||||
}
|
||||
|
||||
message ChatDefaultWallpaper {
|
||||
optional bool isDoodleEnabled = 1;
|
||||
}
|
||||
|
||||
message ChatSolidColorWallpaper {
|
||||
optional string colorLight = 1;
|
||||
optional string colorDark = 2;
|
||||
optional bool isDoodleEnabled = 3;
|
||||
}
|
||||
|
||||
message ChatStockImageWallpaper {
|
||||
optional string stockImageId = 1;
|
||||
optional float dimLevel = 2;
|
||||
}
|
||||
|
||||
message ChatThemeSetting {
|
||||
optional int64 settingTimestampMs = 1;
|
||||
optional bool clearTheme = 2;
|
||||
optional string colorSchemeId = 3;
|
||||
oneof wallpaper {
|
||||
Message.ChatDefaultWallpaper defaultWallpaper = 10;
|
||||
Message.ChatSolidColorWallpaper solidColor = 11;
|
||||
Message.ChatStockImageWallpaper stockImage = 12;
|
||||
Message.ChatCustomImageWallpaper customImage = 13;
|
||||
}
|
||||
}
|
||||
|
||||
message CloudAPIThreadControlNotification {
|
||||
optional CloudAPIThreadControl status = 1;
|
||||
optional int64 senderNotificationTimestampMs = 2;
|
||||
@@ -2742,7 +2637,6 @@ message Message {
|
||||
optional int64 startTime = 5;
|
||||
optional string caption = 6;
|
||||
optional bool isCanceled = 7;
|
||||
optional int64 endTime = 8;
|
||||
}
|
||||
|
||||
message EventMessage {
|
||||
@@ -3325,35 +3219,6 @@ message Message {
|
||||
}
|
||||
}
|
||||
|
||||
message P2PPaymentReminderNotification {
|
||||
optional string reminderId = 1;
|
||||
optional Money amount = 2;
|
||||
optional ReminderFrequency frequency = 3;
|
||||
optional int64 nextReminderTimestamp = 4;
|
||||
optional int64 expiryTimestamp = 5;
|
||||
optional ReminderState state = 6;
|
||||
optional string description = 7;
|
||||
optional string creatorJid = 8;
|
||||
optional string receiverJid = 9;
|
||||
optional string upiId = 10;
|
||||
optional int64 createdTimestamp = 11;
|
||||
enum ReminderFrequency {
|
||||
UNKNOWN_FREQUENCY = 0;
|
||||
WEEKLY = 1;
|
||||
BIWEEKLY = 2;
|
||||
MONTHLY = 3;
|
||||
CUSTOM = 4;
|
||||
}
|
||||
enum ReminderState {
|
||||
UNKNOWN_STATE = 0;
|
||||
ACTIVE = 1;
|
||||
PAUSED = 2;
|
||||
STOPPED = 3;
|
||||
EXPIRED = 4;
|
||||
CANCELLED = 5;
|
||||
}
|
||||
}
|
||||
|
||||
message PaymentExtendedMetadata {
|
||||
optional uint32 type = 1;
|
||||
optional string platform = 2;
|
||||
@@ -3763,7 +3628,6 @@ message Message {
|
||||
optional MemberLabel memberLabel = 27;
|
||||
optional AIMediaCollectionMessage aiMediaCollectionMessage = 28;
|
||||
optional uint32 afterReadDuration = 29;
|
||||
optional Message.ChatThemeSetting chatThemeSetting = 30;
|
||||
enum Type {
|
||||
REVOKE = 0;
|
||||
EPHEMERAL_SETTING = 3;
|
||||
@@ -3793,7 +3657,6 @@ message Message {
|
||||
GROUP_MEMBER_LABEL_CHANGE = 30;
|
||||
AI_MEDIA_COLLECTION_MESSAGE = 31;
|
||||
MESSAGE_UNSCHEDULE = 32;
|
||||
CHAT_THEME_SETTING = 34;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3885,26 +3748,6 @@ message Message {
|
||||
optional bytes axolotlSenderKeyDistributionMessage = 2;
|
||||
}
|
||||
|
||||
message SplitPaymentMessage {
|
||||
optional string splitId = 1;
|
||||
optional Money totalAmount = 2;
|
||||
optional string description = 3;
|
||||
optional string requesterJid = 4;
|
||||
repeated Message.SplitPaymentParticipant participants = 5;
|
||||
optional int64 createdAtMs = 6;
|
||||
optional ContextInfo contextInfo = 17;
|
||||
}
|
||||
|
||||
message SplitPaymentParticipant {
|
||||
optional string jid = 1;
|
||||
optional Money amount = 2;
|
||||
optional SplitPaymentStatus status = 3;
|
||||
enum SplitPaymentStatus {
|
||||
PENDING = 0;
|
||||
PAID = 1;
|
||||
}
|
||||
}
|
||||
|
||||
message StatusNotificationMessage {
|
||||
optional MessageKey responseMessageKey = 1;
|
||||
optional MessageKey originalMessageKey = 2;
|
||||
@@ -4855,7 +4698,6 @@ message StatusAttribution {
|
||||
SHARECHAT = 9;
|
||||
GOOGLE_PHOTOS = 10;
|
||||
SOUNDCLOUD = 11;
|
||||
SHAZAM = 12;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4913,7 +4755,6 @@ message StatusAttribution {
|
||||
LAYOUTS = 8;
|
||||
NEWSLETTER_STATUS = 9;
|
||||
STATUS_CLOSE_SHARING = 10;
|
||||
PAID_PARTNERSHIP = 11;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5425,7 +5266,6 @@ message SyncActionValue {
|
||||
repeated string keywords = 3;
|
||||
optional int32 count = 4;
|
||||
optional bool deleted = 5;
|
||||
repeated string associatedLabelIds = 6;
|
||||
}
|
||||
|
||||
message RecentEmojiWeightsAction {
|
||||
@@ -5943,8 +5783,6 @@ message WebMessageInfo {
|
||||
optional string hsmTag = 79;
|
||||
optional uint64 ephemeralExpirationTimestamp = 80;
|
||||
optional ScheduledMessageMetadata scheduledMessageMetadata = 81;
|
||||
optional string decisionId = 82;
|
||||
repeated string decisionSources = 83;
|
||||
enum BizPrivacyStatus {
|
||||
E2EE = 0;
|
||||
FB = 2;
|
||||
|
||||
Vendored
+5
-377
@@ -28,8 +28,7 @@ export namespace proto {
|
||||
|
||||
enum ADVEncryptionType {
|
||||
E2EE = 0,
|
||||
HOSTED = 1,
|
||||
NON_E2EE = 2
|
||||
HOSTED = 1
|
||||
}
|
||||
|
||||
interface IADVKeyIndexList {
|
||||
@@ -177,8 +176,7 @@ export namespace proto {
|
||||
CREATE_IMAGE = 1,
|
||||
ANIMATE_PHOTO = 2,
|
||||
ANALYZE_FILE = 3,
|
||||
COLLABORATE = 4,
|
||||
OPEN_GREETING_CARD = 5
|
||||
COLLABORATE = 4
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -707,29 +705,6 @@ export namespace proto {
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
enum AISubscriptionRequestType {
|
||||
UNSPECIFIED = 0,
|
||||
THINK_HARD = 1,
|
||||
IMAGE_GEN = 2,
|
||||
VIDEO_GEN = 3
|
||||
}
|
||||
|
||||
interface IAISubscriptionUpsellMetadata {
|
||||
requestType?: (proto.AISubscriptionRequestType|null);
|
||||
}
|
||||
|
||||
class AISubscriptionUpsellMetadata implements IAISubscriptionUpsellMetadata {
|
||||
constructor(p?: proto.IAISubscriptionUpsellMetadata);
|
||||
public requestType?: (proto.AISubscriptionRequestType|null);
|
||||
public static create(properties?: proto.IAISubscriptionUpsellMetadata): proto.AISubscriptionUpsellMetadata;
|
||||
public static encode(m: proto.IAISubscriptionUpsellMetadata, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.AISubscriptionUpsellMetadata;
|
||||
public static fromObject(d: { [k: string]: any }): proto.AISubscriptionUpsellMetadata;
|
||||
public static toObject(m: proto.AISubscriptionUpsellMetadata, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface IAIThreadInfo {
|
||||
serverInfo?: (proto.AIThreadInfo.IAIThreadServerInfo|null);
|
||||
clientInfo?: (proto.AIThreadInfo.IAIThreadClientInfo|null);
|
||||
@@ -1727,8 +1702,6 @@ export namespace proto {
|
||||
botInfrastructureDiagnostics?: (proto.IBotInfrastructureDiagnostics|null);
|
||||
aiMediaCollectionMetadata?: (proto.IAIMediaCollectionMetadata|null);
|
||||
commandMetadata?: (proto.IBotCommandMetadata|null);
|
||||
resolvedToolCallMetadata?: (proto.IBotResolvedToolCallMetadata|null);
|
||||
subscriptionUpsellMetadata?: (proto.IAISubscriptionUpsellMetadata|null);
|
||||
internalMetadata?: (Uint8Array|null);
|
||||
}
|
||||
|
||||
@@ -1772,8 +1745,6 @@ export namespace proto {
|
||||
public botInfrastructureDiagnostics?: (proto.IBotInfrastructureDiagnostics|null);
|
||||
public aiMediaCollectionMetadata?: (proto.IAIMediaCollectionMetadata|null);
|
||||
public commandMetadata?: (proto.IBotCommandMetadata|null);
|
||||
public resolvedToolCallMetadata?: (proto.IBotResolvedToolCallMetadata|null);
|
||||
public subscriptionUpsellMetadata?: (proto.IAISubscriptionUpsellMetadata|null);
|
||||
public internalMetadata?: (Uint8Array|null);
|
||||
public static create(properties?: proto.IBotMetadata): proto.BotMetadata;
|
||||
public static encode(m: proto.IBotMetadata, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
@@ -2323,24 +2294,6 @@ export namespace proto {
|
||||
}
|
||||
}
|
||||
|
||||
interface IBotResolvedToolCallMetadata {
|
||||
toolCallId?: (string|null);
|
||||
resolutionDataSerialized?: (string|null);
|
||||
}
|
||||
|
||||
class BotResolvedToolCallMetadata implements IBotResolvedToolCallMetadata {
|
||||
constructor(p?: proto.IBotResolvedToolCallMetadata);
|
||||
public toolCallId?: (string|null);
|
||||
public resolutionDataSerialized?: (string|null);
|
||||
public static create(properties?: proto.IBotResolvedToolCallMetadata): proto.BotResolvedToolCallMetadata;
|
||||
public static encode(m: proto.IBotResolvedToolCallMetadata, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.BotResolvedToolCallMetadata;
|
||||
public static fromObject(d: { [k: string]: any }): proto.BotResolvedToolCallMetadata;
|
||||
public static toObject(m: proto.BotResolvedToolCallMetadata, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface IBotSessionMetadata {
|
||||
sessionId?: (string|null);
|
||||
sessionSource?: (proto.BotSessionSource|null);
|
||||
@@ -2929,7 +2882,6 @@ export namespace proto {
|
||||
preacksCount?: (number|null);
|
||||
processingQueueSize?: (number|null);
|
||||
pairedPeripherals?: (string[]|null);
|
||||
testIsolationId?: (Uint8Array|null);
|
||||
}
|
||||
|
||||
class ClientPayload implements IClientPayload {
|
||||
@@ -2969,7 +2921,6 @@ export namespace proto {
|
||||
public preacksCount?: (number|null);
|
||||
public processingQueueSize?: (number|null);
|
||||
public pairedPeripherals: string[];
|
||||
public testIsolationId?: (Uint8Array|null);
|
||||
public static create(properties?: proto.IClientPayload): proto.ClientPayload;
|
||||
public static encode(m: proto.IClientPayload, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.ClientPayload;
|
||||
@@ -3132,7 +3083,6 @@ export namespace proto {
|
||||
deviceExpId?: (string|null);
|
||||
deviceType?: (proto.ClientPayload.UserAgent.DeviceType|null);
|
||||
deviceModelType?: (string|null);
|
||||
distributionChannel?: (proto.ClientPayload.UserAgent.DistributionChannel|null);
|
||||
}
|
||||
|
||||
class UserAgent implements IUserAgent {
|
||||
@@ -3153,7 +3103,6 @@ export namespace proto {
|
||||
public deviceExpId?: (string|null);
|
||||
public deviceType?: (proto.ClientPayload.UserAgent.DeviceType|null);
|
||||
public deviceModelType?: (string|null);
|
||||
public distributionChannel?: (proto.ClientPayload.UserAgent.DistributionChannel|null);
|
||||
public static create(properties?: proto.ClientPayload.IUserAgent): proto.ClientPayload.UserAgent;
|
||||
public static encode(m: proto.ClientPayload.IUserAgent, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.ClientPayload.UserAgent;
|
||||
@@ -3197,13 +3146,6 @@ export namespace proto {
|
||||
VR = 4
|
||||
}
|
||||
|
||||
enum DistributionChannel {
|
||||
APPSTORE = 0,
|
||||
WEBSITE = 1,
|
||||
TESTFLIGHT = 2,
|
||||
INTERNAL = 3
|
||||
}
|
||||
|
||||
enum Platform {
|
||||
ANDROID = 0,
|
||||
IOS = 1,
|
||||
@@ -3469,9 +3411,6 @@ export namespace proto {
|
||||
mediaDomainInfo?: (proto.IMediaDomainInfo|null);
|
||||
partiallySelectedContent?: (proto.ContextInfo.IPartiallySelectedContent|null);
|
||||
afterReadDuration?: (number|null);
|
||||
crossAppSource?: (proto.ContextInfo.CrossAppSource|null);
|
||||
businessInteractionPills?: (proto.ContextInfo.IBusinessInteractionPills|null);
|
||||
posterStatusId?: (string|null);
|
||||
}
|
||||
|
||||
class ContextInfo implements IContextInfo {
|
||||
@@ -3535,9 +3474,6 @@ export namespace proto {
|
||||
public mediaDomainInfo?: (proto.IMediaDomainInfo|null);
|
||||
public partiallySelectedContent?: (proto.ContextInfo.IPartiallySelectedContent|null);
|
||||
public afterReadDuration?: (number|null);
|
||||
public crossAppSource?: (proto.ContextInfo.CrossAppSource|null);
|
||||
public businessInteractionPills?: (proto.ContextInfo.IBusinessInteractionPills|null);
|
||||
public posterStatusId?: (string|null);
|
||||
public static create(properties?: proto.IContextInfo): proto.ContextInfo;
|
||||
public static encode(m: proto.IContextInfo, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.ContextInfo;
|
||||
@@ -3580,70 +3516,6 @@ export namespace proto {
|
||||
}
|
||||
}
|
||||
|
||||
interface IBusinessInteractionPills {
|
||||
businessJid?: (string|null);
|
||||
pills?: (proto.ContextInfo.BusinessInteractionPills.IPill[]|null);
|
||||
entryPoint?: (proto.ContextInfo.BusinessInteractionPills.EntryPoint|null);
|
||||
}
|
||||
|
||||
class BusinessInteractionPills implements IBusinessInteractionPills {
|
||||
constructor(p?: proto.ContextInfo.IBusinessInteractionPills);
|
||||
public businessJid?: (string|null);
|
||||
public pills: proto.ContextInfo.BusinessInteractionPills.IPill[];
|
||||
public entryPoint?: (proto.ContextInfo.BusinessInteractionPills.EntryPoint|null);
|
||||
public static create(properties?: proto.ContextInfo.IBusinessInteractionPills): proto.ContextInfo.BusinessInteractionPills;
|
||||
public static encode(m: proto.ContextInfo.IBusinessInteractionPills, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.ContextInfo.BusinessInteractionPills;
|
||||
public static fromObject(d: { [k: string]: any }): proto.ContextInfo.BusinessInteractionPills;
|
||||
public static toObject(m: proto.ContextInfo.BusinessInteractionPills, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
namespace BusinessInteractionPills {
|
||||
|
||||
enum EntryPoint {
|
||||
ENTRY_POINT_UNKNOWN = 0,
|
||||
P2P_LINK_SHARE = 1,
|
||||
CONTACT_CARD_SHARING = 2,
|
||||
PHONE_NUMBER = 3,
|
||||
STATUS = 4,
|
||||
IN_THREAD_CONTEXT_CARD = 5
|
||||
}
|
||||
|
||||
interface IPill {
|
||||
pillType?: (proto.ContextInfo.BusinessInteractionPills.PillType|null);
|
||||
actionUrl?: (string|null);
|
||||
}
|
||||
|
||||
class Pill implements IPill {
|
||||
constructor(p?: proto.ContextInfo.BusinessInteractionPills.IPill);
|
||||
public pillType?: (proto.ContextInfo.BusinessInteractionPills.PillType|null);
|
||||
public actionUrl?: (string|null);
|
||||
public static create(properties?: proto.ContextInfo.BusinessInteractionPills.IPill): proto.ContextInfo.BusinessInteractionPills.Pill;
|
||||
public static encode(m: proto.ContextInfo.BusinessInteractionPills.IPill, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.ContextInfo.BusinessInteractionPills.Pill;
|
||||
public static fromObject(d: { [k: string]: any }): proto.ContextInfo.BusinessInteractionPills.Pill;
|
||||
public static toObject(m: proto.ContextInfo.BusinessInteractionPills.Pill, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
enum PillType {
|
||||
UNKNOWN = 0,
|
||||
VIEW_BUSINESS = 1,
|
||||
CHAT = 2,
|
||||
CALL = 3,
|
||||
CATALOG = 4,
|
||||
CHANNEL = 5,
|
||||
BOOK_APPOINTMENT = 6,
|
||||
OFFERS = 7,
|
||||
BESTSELLERS = 8,
|
||||
MENU = 9,
|
||||
ABOUT = 10
|
||||
}
|
||||
}
|
||||
|
||||
interface IBusinessMessageForwardInfo {
|
||||
businessOwnerJid?: (string|null);
|
||||
}
|
||||
@@ -3660,12 +3532,6 @@ export namespace proto {
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
enum CrossAppSource {
|
||||
CROSS_APP_SOURCE_UNKNOWN = 0,
|
||||
CROSS_APP_SOURCE_INSTAGRAM = 1,
|
||||
CROSS_APP_SOURCE_FACEBOOK = 2
|
||||
}
|
||||
|
||||
interface IDataSharingContext {
|
||||
showMmDisclosure?: (boolean|null);
|
||||
encryptedSignalTokenConsented?: (string|null);
|
||||
@@ -4055,7 +3921,6 @@ export namespace proto {
|
||||
isMarketingMessageThread?: (boolean|null);
|
||||
isSenderNewAccount?: (boolean|null);
|
||||
afterReadDuration?: (number|null);
|
||||
isSenderSuspicious?: (boolean|null);
|
||||
}
|
||||
|
||||
class Conversation implements IConversation {
|
||||
@@ -4117,7 +3982,6 @@ export namespace proto {
|
||||
public isMarketingMessageThread?: (boolean|null);
|
||||
public isSenderNewAccount?: (boolean|null);
|
||||
public afterReadDuration?: (number|null);
|
||||
public isSenderSuspicious?: (boolean|null);
|
||||
public static create(properties?: proto.IConversation): proto.Conversation;
|
||||
public static encode(m: proto.IConversation, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Conversation;
|
||||
@@ -5861,8 +5725,6 @@ export namespace proto {
|
||||
pollAddOptionMessage?: (proto.Message.IPollAddOptionMessage|null);
|
||||
eventInviteMessage?: (proto.Message.IEventInviteMessage|null);
|
||||
groupRootKeyShare?: (proto.IGroupRootKeyShare|null);
|
||||
p2PPaymentReminderNotification?: (proto.Message.IP2PPaymentReminderNotification|null);
|
||||
splitPaymentMessage?: (proto.Message.ISplitPaymentMessage|null);
|
||||
}
|
||||
|
||||
class Message implements IMessage {
|
||||
@@ -5970,8 +5832,6 @@ export namespace proto {
|
||||
public pollAddOptionMessage?: (proto.Message.IPollAddOptionMessage|null);
|
||||
public eventInviteMessage?: (proto.Message.IEventInviteMessage|null);
|
||||
public groupRootKeyShare?: (proto.IGroupRootKeyShare|null);
|
||||
public p2PPaymentReminderNotification?: (proto.Message.IP2PPaymentReminderNotification|null);
|
||||
public splitPaymentMessage?: (proto.Message.ISplitPaymentMessage|null);
|
||||
public static create(properties?: proto.IMessage): proto.Message;
|
||||
public static encode(m: proto.IMessage, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message;
|
||||
@@ -6480,113 +6340,6 @@ export namespace proto {
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface IChatCustomImageWallpaper {
|
||||
directPath?: (string|null);
|
||||
mediaKey?: (Uint8Array|null);
|
||||
fileEncSha256?: (Uint8Array|null);
|
||||
fileSha256?: (Uint8Array|null);
|
||||
dimLevel?: (number|null);
|
||||
}
|
||||
|
||||
class ChatCustomImageWallpaper implements IChatCustomImageWallpaper {
|
||||
constructor(p?: proto.Message.IChatCustomImageWallpaper);
|
||||
public directPath?: (string|null);
|
||||
public mediaKey?: (Uint8Array|null);
|
||||
public fileEncSha256?: (Uint8Array|null);
|
||||
public fileSha256?: (Uint8Array|null);
|
||||
public dimLevel?: (number|null);
|
||||
public static create(properties?: proto.Message.IChatCustomImageWallpaper): proto.Message.ChatCustomImageWallpaper;
|
||||
public static encode(m: proto.Message.IChatCustomImageWallpaper, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.ChatCustomImageWallpaper;
|
||||
public static fromObject(d: { [k: string]: any }): proto.Message.ChatCustomImageWallpaper;
|
||||
public static toObject(m: proto.Message.ChatCustomImageWallpaper, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface IChatDefaultWallpaper {
|
||||
isDoodleEnabled?: (boolean|null);
|
||||
}
|
||||
|
||||
class ChatDefaultWallpaper implements IChatDefaultWallpaper {
|
||||
constructor(p?: proto.Message.IChatDefaultWallpaper);
|
||||
public isDoodleEnabled?: (boolean|null);
|
||||
public static create(properties?: proto.Message.IChatDefaultWallpaper): proto.Message.ChatDefaultWallpaper;
|
||||
public static encode(m: proto.Message.IChatDefaultWallpaper, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.ChatDefaultWallpaper;
|
||||
public static fromObject(d: { [k: string]: any }): proto.Message.ChatDefaultWallpaper;
|
||||
public static toObject(m: proto.Message.ChatDefaultWallpaper, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface IChatSolidColorWallpaper {
|
||||
colorLight?: (string|null);
|
||||
colorDark?: (string|null);
|
||||
isDoodleEnabled?: (boolean|null);
|
||||
}
|
||||
|
||||
class ChatSolidColorWallpaper implements IChatSolidColorWallpaper {
|
||||
constructor(p?: proto.Message.IChatSolidColorWallpaper);
|
||||
public colorLight?: (string|null);
|
||||
public colorDark?: (string|null);
|
||||
public isDoodleEnabled?: (boolean|null);
|
||||
public static create(properties?: proto.Message.IChatSolidColorWallpaper): proto.Message.ChatSolidColorWallpaper;
|
||||
public static encode(m: proto.Message.IChatSolidColorWallpaper, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.ChatSolidColorWallpaper;
|
||||
public static fromObject(d: { [k: string]: any }): proto.Message.ChatSolidColorWallpaper;
|
||||
public static toObject(m: proto.Message.ChatSolidColorWallpaper, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface IChatStockImageWallpaper {
|
||||
stockImageId?: (string|null);
|
||||
dimLevel?: (number|null);
|
||||
}
|
||||
|
||||
class ChatStockImageWallpaper implements IChatStockImageWallpaper {
|
||||
constructor(p?: proto.Message.IChatStockImageWallpaper);
|
||||
public stockImageId?: (string|null);
|
||||
public dimLevel?: (number|null);
|
||||
public static create(properties?: proto.Message.IChatStockImageWallpaper): proto.Message.ChatStockImageWallpaper;
|
||||
public static encode(m: proto.Message.IChatStockImageWallpaper, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.ChatStockImageWallpaper;
|
||||
public static fromObject(d: { [k: string]: any }): proto.Message.ChatStockImageWallpaper;
|
||||
public static toObject(m: proto.Message.ChatStockImageWallpaper, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface IChatThemeSetting {
|
||||
settingTimestampMs?: (number|Long|null);
|
||||
clearTheme?: (boolean|null);
|
||||
colorSchemeId?: (string|null);
|
||||
defaultWallpaper?: (proto.Message.IChatDefaultWallpaper|null);
|
||||
solidColor?: (proto.Message.IChatSolidColorWallpaper|null);
|
||||
stockImage?: (proto.Message.IChatStockImageWallpaper|null);
|
||||
customImage?: (proto.Message.IChatCustomImageWallpaper|null);
|
||||
}
|
||||
|
||||
class ChatThemeSetting implements IChatThemeSetting {
|
||||
constructor(p?: proto.Message.IChatThemeSetting);
|
||||
public settingTimestampMs?: (number|Long|null);
|
||||
public clearTheme?: (boolean|null);
|
||||
public colorSchemeId?: (string|null);
|
||||
public defaultWallpaper?: (proto.Message.IChatDefaultWallpaper|null);
|
||||
public solidColor?: (proto.Message.IChatSolidColorWallpaper|null);
|
||||
public stockImage?: (proto.Message.IChatStockImageWallpaper|null);
|
||||
public customImage?: (proto.Message.IChatCustomImageWallpaper|null);
|
||||
public wallpaper?: ("defaultWallpaper"|"solidColor"|"stockImage"|"customImage");
|
||||
public static create(properties?: proto.Message.IChatThemeSetting): proto.Message.ChatThemeSetting;
|
||||
public static encode(m: proto.Message.IChatThemeSetting, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.ChatThemeSetting;
|
||||
public static fromObject(d: { [k: string]: any }): proto.Message.ChatThemeSetting;
|
||||
public static toObject(m: proto.Message.ChatThemeSetting, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface ICloudAPIThreadControlNotification {
|
||||
status?: (proto.Message.CloudAPIThreadControlNotification.CloudAPIThreadControl|null);
|
||||
senderNotificationTimestampMs?: (number|Long|null);
|
||||
@@ -6891,7 +6644,6 @@ export namespace proto {
|
||||
startTime?: (number|Long|null);
|
||||
caption?: (string|null);
|
||||
isCanceled?: (boolean|null);
|
||||
endTime?: (number|Long|null);
|
||||
}
|
||||
|
||||
class EventInviteMessage implements IEventInviteMessage {
|
||||
@@ -6903,7 +6655,6 @@ export namespace proto {
|
||||
public startTime?: (number|Long|null);
|
||||
public caption?: (string|null);
|
||||
public isCanceled?: (boolean|null);
|
||||
public endTime?: (number|Long|null);
|
||||
public static create(properties?: proto.Message.IEventInviteMessage): proto.Message.EventInviteMessage;
|
||||
public static encode(m: proto.Message.IEventInviteMessage, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.EventInviteMessage;
|
||||
@@ -8424,62 +8175,6 @@ export namespace proto {
|
||||
}
|
||||
}
|
||||
|
||||
interface IP2PPaymentReminderNotification {
|
||||
reminderId?: (string|null);
|
||||
amount?: (proto.IMoney|null);
|
||||
frequency?: (proto.Message.P2PPaymentReminderNotification.ReminderFrequency|null);
|
||||
nextReminderTimestamp?: (number|Long|null);
|
||||
expiryTimestamp?: (number|Long|null);
|
||||
state?: (proto.Message.P2PPaymentReminderNotification.ReminderState|null);
|
||||
description?: (string|null);
|
||||
creatorJid?: (string|null);
|
||||
receiverJid?: (string|null);
|
||||
upiId?: (string|null);
|
||||
createdTimestamp?: (number|Long|null);
|
||||
}
|
||||
|
||||
class P2PPaymentReminderNotification implements IP2PPaymentReminderNotification {
|
||||
constructor(p?: proto.Message.IP2PPaymentReminderNotification);
|
||||
public reminderId?: (string|null);
|
||||
public amount?: (proto.IMoney|null);
|
||||
public frequency?: (proto.Message.P2PPaymentReminderNotification.ReminderFrequency|null);
|
||||
public nextReminderTimestamp?: (number|Long|null);
|
||||
public expiryTimestamp?: (number|Long|null);
|
||||
public state?: (proto.Message.P2PPaymentReminderNotification.ReminderState|null);
|
||||
public description?: (string|null);
|
||||
public creatorJid?: (string|null);
|
||||
public receiverJid?: (string|null);
|
||||
public upiId?: (string|null);
|
||||
public createdTimestamp?: (number|Long|null);
|
||||
public static create(properties?: proto.Message.IP2PPaymentReminderNotification): proto.Message.P2PPaymentReminderNotification;
|
||||
public static encode(m: proto.Message.IP2PPaymentReminderNotification, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.P2PPaymentReminderNotification;
|
||||
public static fromObject(d: { [k: string]: any }): proto.Message.P2PPaymentReminderNotification;
|
||||
public static toObject(m: proto.Message.P2PPaymentReminderNotification, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
namespace P2PPaymentReminderNotification {
|
||||
|
||||
enum ReminderFrequency {
|
||||
UNKNOWN_FREQUENCY = 0,
|
||||
WEEKLY = 1,
|
||||
BIWEEKLY = 2,
|
||||
MONTHLY = 3,
|
||||
CUSTOM = 4
|
||||
}
|
||||
|
||||
enum ReminderState {
|
||||
UNKNOWN_STATE = 0,
|
||||
ACTIVE = 1,
|
||||
PAUSED = 2,
|
||||
STOPPED = 3,
|
||||
EXPIRED = 4,
|
||||
CANCELLED = 5
|
||||
}
|
||||
}
|
||||
|
||||
interface IPaymentExtendedMetadata {
|
||||
type?: (number|null);
|
||||
platform?: (string|null);
|
||||
@@ -9628,7 +9323,6 @@ export namespace proto {
|
||||
memberLabel?: (proto.IMemberLabel|null);
|
||||
aiMediaCollectionMessage?: (proto.IAIMediaCollectionMessage|null);
|
||||
afterReadDuration?: (number|null);
|
||||
chatThemeSetting?: (proto.Message.IChatThemeSetting|null);
|
||||
}
|
||||
|
||||
class ProtocolMessage implements IProtocolMessage {
|
||||
@@ -9659,7 +9353,6 @@ export namespace proto {
|
||||
public memberLabel?: (proto.IMemberLabel|null);
|
||||
public aiMediaCollectionMessage?: (proto.IAIMediaCollectionMessage|null);
|
||||
public afterReadDuration?: (number|null);
|
||||
public chatThemeSetting?: (proto.Message.IChatThemeSetting|null);
|
||||
public static create(properties?: proto.Message.IProtocolMessage): proto.Message.ProtocolMessage;
|
||||
public static encode(m: proto.Message.IProtocolMessage, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.ProtocolMessage;
|
||||
@@ -9699,8 +9392,7 @@ export namespace proto {
|
||||
AI_QUERY_FANOUT = 29,
|
||||
GROUP_MEMBER_LABEL_CHANGE = 30,
|
||||
AI_MEDIA_COLLECTION_MESSAGE = 31,
|
||||
MESSAGE_UNSCHEDULE = 32,
|
||||
CHAT_THEME_SETTING = 34
|
||||
MESSAGE_UNSCHEDULE = 32
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9952,62 +9644,6 @@ export namespace proto {
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface ISplitPaymentMessage {
|
||||
splitId?: (string|null);
|
||||
totalAmount?: (proto.IMoney|null);
|
||||
description?: (string|null);
|
||||
requesterJid?: (string|null);
|
||||
participants?: (proto.Message.ISplitPaymentParticipant[]|null);
|
||||
createdAtMs?: (number|Long|null);
|
||||
contextInfo?: (proto.IContextInfo|null);
|
||||
}
|
||||
|
||||
class SplitPaymentMessage implements ISplitPaymentMessage {
|
||||
constructor(p?: proto.Message.ISplitPaymentMessage);
|
||||
public splitId?: (string|null);
|
||||
public totalAmount?: (proto.IMoney|null);
|
||||
public description?: (string|null);
|
||||
public requesterJid?: (string|null);
|
||||
public participants: proto.Message.ISplitPaymentParticipant[];
|
||||
public createdAtMs?: (number|Long|null);
|
||||
public contextInfo?: (proto.IContextInfo|null);
|
||||
public static create(properties?: proto.Message.ISplitPaymentMessage): proto.Message.SplitPaymentMessage;
|
||||
public static encode(m: proto.Message.ISplitPaymentMessage, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.SplitPaymentMessage;
|
||||
public static fromObject(d: { [k: string]: any }): proto.Message.SplitPaymentMessage;
|
||||
public static toObject(m: proto.Message.SplitPaymentMessage, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
interface ISplitPaymentParticipant {
|
||||
jid?: (string|null);
|
||||
amount?: (proto.IMoney|null);
|
||||
status?: (proto.Message.SplitPaymentParticipant.SplitPaymentStatus|null);
|
||||
}
|
||||
|
||||
class SplitPaymentParticipant implements ISplitPaymentParticipant {
|
||||
constructor(p?: proto.Message.ISplitPaymentParticipant);
|
||||
public jid?: (string|null);
|
||||
public amount?: (proto.IMoney|null);
|
||||
public status?: (proto.Message.SplitPaymentParticipant.SplitPaymentStatus|null);
|
||||
public static create(properties?: proto.Message.ISplitPaymentParticipant): proto.Message.SplitPaymentParticipant;
|
||||
public static encode(m: proto.Message.ISplitPaymentParticipant, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.Message.SplitPaymentParticipant;
|
||||
public static fromObject(d: { [k: string]: any }): proto.Message.SplitPaymentParticipant;
|
||||
public static toObject(m: proto.Message.SplitPaymentParticipant, o?: $protobuf.IConversionOptions): { [k: string]: any };
|
||||
public toJSON(): { [k: string]: any };
|
||||
public static getTypeUrl(typeUrlPrefix?: string): string;
|
||||
}
|
||||
|
||||
namespace SplitPaymentParticipant {
|
||||
|
||||
enum SplitPaymentStatus {
|
||||
PENDING = 0,
|
||||
PAID = 1
|
||||
}
|
||||
}
|
||||
|
||||
interface IStatusNotificationMessage {
|
||||
responseMessageKey?: (proto.IMessageKey|null);
|
||||
originalMessageKey?: (proto.IMessageKey|null);
|
||||
@@ -12362,8 +11998,7 @@ export namespace proto {
|
||||
APPLE_MUSIC = 8,
|
||||
SHARECHAT = 9,
|
||||
GOOGLE_PHOTOS = 10,
|
||||
SOUNDCLOUD = 11,
|
||||
SHAZAM = 12
|
||||
SOUNDCLOUD = 11
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12497,8 +12132,7 @@ export namespace proto {
|
||||
AI_CREATED = 7,
|
||||
LAYOUTS = 8,
|
||||
NEWSLETTER_STATUS = 9,
|
||||
STATUS_CLOSE_SHARING = 10,
|
||||
PAID_PARTNERSHIP = 11
|
||||
STATUS_CLOSE_SHARING = 10
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14035,7 +13669,6 @@ export namespace proto {
|
||||
keywords?: (string[]|null);
|
||||
count?: (number|null);
|
||||
deleted?: (boolean|null);
|
||||
associatedLabelIds?: (string[]|null);
|
||||
}
|
||||
|
||||
class QuickReplyAction implements IQuickReplyAction {
|
||||
@@ -14045,7 +13678,6 @@ export namespace proto {
|
||||
public keywords: string[];
|
||||
public count?: (number|null);
|
||||
public deleted?: (boolean|null);
|
||||
public associatedLabelIds: string[];
|
||||
public static create(properties?: proto.SyncActionValue.IQuickReplyAction): proto.SyncActionValue.QuickReplyAction;
|
||||
public static encode(m: proto.SyncActionValue.IQuickReplyAction, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.SyncActionValue.QuickReplyAction;
|
||||
@@ -15327,8 +14959,6 @@ export namespace proto {
|
||||
hsmTag?: (string|null);
|
||||
ephemeralExpirationTimestamp?: (number|Long|null);
|
||||
scheduledMessageMetadata?: (proto.IScheduledMessageMetadata|null);
|
||||
decisionId?: (string|null);
|
||||
decisionSources?: (string[]|null);
|
||||
}
|
||||
|
||||
class WebMessageInfo implements IWebMessageInfo {
|
||||
@@ -15403,8 +15033,6 @@ export namespace proto {
|
||||
public hsmTag?: (string|null);
|
||||
public ephemeralExpirationTimestamp?: (number|Long|null);
|
||||
public scheduledMessageMetadata?: (proto.IScheduledMessageMetadata|null);
|
||||
public decisionId?: (string|null);
|
||||
public decisionSources: string[];
|
||||
public static create(properties?: proto.IWebMessageInfo): proto.WebMessageInfo;
|
||||
public static encode(m: proto.IWebMessageInfo, w?: $protobuf.Writer): $protobuf.Writer;
|
||||
public static decode(r: ($protobuf.Reader|Uint8Array), l?: number): proto.WebMessageInfo;
|
||||
|
||||
-2549
File diff suppressed because it is too large
Load Diff
@@ -1 +1 @@
|
||||
{"version":[2,3000,1038585534]}
|
||||
{"version":[2,3000,1038167900]}
|
||||
|
||||
@@ -142,6 +142,8 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = {
|
||||
getMessage: async () => undefined,
|
||||
cachedGroupMetadata: async () => undefined,
|
||||
makeSignalRepository: makeLibSignalRepository,
|
||||
// Circuit breaker configuration
|
||||
enableCircuitBreaker: true,
|
||||
// Listener limits (memory leak prevention)
|
||||
// WebSocket: 8 core events (open, close, error, message, ping, pong, upgrade, unexpected-response)
|
||||
// + 10 dynamic listeners (reconnect handlers, custom events)
|
||||
|
||||
+68
-92
@@ -6,6 +6,7 @@ import type { LIDMapping, SignalAuthState, SignalKeyStoreWithTransaction } from
|
||||
import type { BaileysEventEmitter } from '../Types/Events'
|
||||
import type { SignalRepositoryWithLIDStore } from '../Types/Signal'
|
||||
import { generateSignalPubKey } from '../Utils'
|
||||
import { CircuitBreaker } from '../Utils/circuit-breaker.js'
|
||||
import type { ILogger } from '../Utils/logger'
|
||||
import { metrics } from '../Utils/prometheus-metrics.js'
|
||||
import { isAnyLidUser, isAnyPnUser, jidDecode, transferDevice, WAJIDDomains } from '../WABinary'
|
||||
@@ -66,6 +67,8 @@ export interface IdentitySaveResult {
|
||||
export interface LibSignalRepositoryOptions {
|
||||
/** Event emitter for broadcasting identity changes */
|
||||
ev?: BaileysEventEmitter
|
||||
/** Circuit breaker for prekey operations (optional) */
|
||||
preKeyCircuitBreaker?: CircuitBreaker
|
||||
}
|
||||
|
||||
// ============================================
|
||||
@@ -253,7 +256,7 @@ export function makeLibSignalRepository(
|
||||
pnToLIDFunc?: (jids: string[]) => Promise<LIDMapping[] | undefined>,
|
||||
options?: LibSignalRepositoryOptions
|
||||
): SignalRepositoryWithLIDStore {
|
||||
const { ev } = options || {}
|
||||
const { ev, preKeyCircuitBreaker } = options || {}
|
||||
const lidMapping = new LIDMappingStore(auth.keys as SignalKeyStoreWithTransaction, logger, pnToLIDFunc)
|
||||
|
||||
// Identity key cache to avoid repeated storage reads
|
||||
@@ -274,7 +277,7 @@ export function makeLibSignalRepository(
|
||||
cacheMetricsInterval.unref()
|
||||
}
|
||||
|
||||
const storage = signalStorage(auth, lidMapping, identityKeyCache, ev, logger)
|
||||
const storage = signalStorage(auth, lidMapping, identityKeyCache, ev, preKeyCircuitBreaker, logger)
|
||||
|
||||
const parsedKeys = auth.keys as SignalKeyStoreWithTransaction
|
||||
const migratedSessionCache = new LRUCache<string, true>({
|
||||
@@ -299,6 +302,25 @@ export function makeLibSignalRepository(
|
||||
// Promise instead of each spawning their own DB transactions.
|
||||
const migrationInFlight = new Map<string, Promise<{ migrated: number; skipped: number; total: number }>>()
|
||||
|
||||
// Resolve PN JID to its canonical LID JID for transaction locking.
|
||||
// This prevents PN/LID race conditions where concurrent operations for the
|
||||
// same logical contact acquire different mutex locks because one uses PN
|
||||
// and the other uses LID. (Aligned with WABA behavior — all operations use LID internally.)
|
||||
const resolveCanonicalJid = async(jid: string): Promise<string> => {
|
||||
if (isAnyLidUser(jid)) {
|
||||
return jid
|
||||
}
|
||||
|
||||
if (isAnyPnUser(jid)) {
|
||||
const lid = await lidMapping.getLIDForPN(jid)
|
||||
if (lid) {
|
||||
return lid
|
||||
}
|
||||
}
|
||||
|
||||
return jid
|
||||
}
|
||||
|
||||
const repository: SignalRepositoryWithLIDStore = {
|
||||
decryptGroupMessage({ group, authorJid, msg }) {
|
||||
const senderName = jidToSignalSenderKeyName(group, authorJid)
|
||||
@@ -341,14 +363,6 @@ export function makeLibSignalRepository(
|
||||
},
|
||||
async decryptMessage({ jid, type, ciphertext }) {
|
||||
const addr = jidToSignalProtocolAddress(jid)
|
||||
// Wire address = the EXACT key that signalStorage's loadSession/storeSession
|
||||
// uses internally (it canonicalizes PN→LID via resolveLIDSignalAddress).
|
||||
// We MUST lock on this same key — locking on the raw `jid` lets two
|
||||
// concurrent ops for the same logical contact (one arriving as PN, one as
|
||||
// LID) acquire DIFFERENT mutex slots and interleave on the same session
|
||||
// record, corrupting the ratchet → perpetual `Bad MAC` on own DSM after
|
||||
// the WhatsApp PN→LID DSM rollout.
|
||||
const wireAddr = await resolveLIDSignalAddress(addr.toString(), lidMapping)
|
||||
const session = new libsignal.SessionCipher(storage, addr)
|
||||
|
||||
async function doDecrypt() {
|
||||
@@ -371,9 +385,13 @@ export function makeLibSignalRepository(
|
||||
'Identity key changed - contact may have reinstalled WhatsApp, session will be re-established'
|
||||
)
|
||||
|
||||
// (No circuit breaker to reset — bounded-retry is stateless,
|
||||
// each retry is independent so identity-change recovery
|
||||
// happens automatically on the next operation.)
|
||||
// Reset prekey circuit breaker since we identified the cause
|
||||
// Reset regardless of state (could be open, half-open, or closed with accumulated failures)
|
||||
// eslint-disable-next-line max-depth
|
||||
if (preKeyCircuitBreaker) {
|
||||
preKeyCircuitBreaker.reset()
|
||||
logger.debug({ jid }, 'Reset prekey circuit breaker after identity key change detection')
|
||||
}
|
||||
} else if (saveResult.isNew) {
|
||||
logger.debug(
|
||||
{ jid, addr: addrStr, fingerprint: saveResult.currentFingerprint },
|
||||
@@ -397,23 +415,24 @@ export function makeLibSignalRepository(
|
||||
return result
|
||||
}
|
||||
|
||||
// Use canonical JID (PN→LID resolved) as transaction key to prevent
|
||||
// PN/LID race conditions on the same logical session.
|
||||
const canonicalJid = await resolveCanonicalJid(jid)
|
||||
return parsedKeys.transaction(async () => {
|
||||
return await doDecrypt()
|
||||
}, wireAddr)
|
||||
}, canonicalJid)
|
||||
},
|
||||
|
||||
async encryptMessage({ jid, data }) {
|
||||
const addr = jidToSignalProtocolAddress(jid)
|
||||
// See decryptMessage above for the rationale: lock on the wire address,
|
||||
// not the raw JID, so concurrent PN/LID ops on the same session serialize.
|
||||
const wireAddr = await resolveLIDSignalAddress(addr.toString(), lidMapping)
|
||||
const cipher = new libsignal.SessionCipher(storage, addr)
|
||||
|
||||
const canonicalJid = await resolveCanonicalJid(jid)
|
||||
return parsedKeys.transaction(async () => {
|
||||
const { type: sigType, body } = await cipher.encrypt(data)
|
||||
const type = sigType === 3 ? 'pkmsg' : 'msg'
|
||||
return { type, ciphertext: Buffer.from(body, 'binary') }
|
||||
}, wireAddr)
|
||||
}, canonicalJid)
|
||||
},
|
||||
|
||||
async encryptGroupMessage({ group, meId, data }) {
|
||||
@@ -441,16 +460,10 @@ export function makeLibSignalRepository(
|
||||
|
||||
async injectE2ESession({ jid, session }) {
|
||||
logger.trace({ jid }, 'injecting E2EE session')
|
||||
const addr = jidToSignalProtocolAddress(jid)
|
||||
// Same wire-address locking as encrypt/decrypt — `SessionBuilder.initOutgoing`
|
||||
// writes via `storeSession` which canonicalizes, so the lock key must be
|
||||
// the wire address for the mutex to actually serialize concurrent ops on
|
||||
// the same logical session.
|
||||
const wireAddr = await resolveLIDSignalAddress(addr.toString(), lidMapping)
|
||||
const cipher = new libsignal.SessionBuilder(storage, addr)
|
||||
const cipher = new libsignal.SessionBuilder(storage, jidToSignalProtocolAddress(jid))
|
||||
return parsedKeys.transaction(async () => {
|
||||
await cipher.initOutgoing(session)
|
||||
}, wireAddr)
|
||||
}, jid)
|
||||
},
|
||||
jidToSignalProtocolAddress(jid) {
|
||||
return jidToSignalProtocolAddress(jid).toString()
|
||||
@@ -727,65 +740,6 @@ const jidToSignalSenderKeyName = (group: string, user: string): SenderKeyName =>
|
||||
return new SenderKeyName(group, jidToSignalProtocolAddress(user))
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a Signal protocol address (string form `signalUser.device`, where
|
||||
* `signalUser` is `user` for the WhatsApp domain or `user_domainType` for any
|
||||
* other domain — e.g. `12345_1.0` for a LID device 0) to its canonical *wire*
|
||||
* address — the exact key that `loadSession`/`storeSession` uses when
|
||||
* reading/writing `keys.get('session', [...])`.
|
||||
*
|
||||
* This is the SAME key the storage layer uses internally, so callers that need
|
||||
* a transaction lock around session mutations MUST lock on this resolved
|
||||
* address (not the raw JID), otherwise the lock key and storage key drift
|
||||
* apart and concurrent ops on the same session interleave → ratchet
|
||||
* corruption → perpetual `Bad MAC` errors.
|
||||
*
|
||||
* Behavior: if the input already targets a LID domain, returns it unchanged
|
||||
* (cheap fast path — no awaits). Otherwise looks up the PN→LID mapping; if
|
||||
* found, returns the LID-form Signal address. If no mapping exists, returns
|
||||
* the input unchanged.
|
||||
*
|
||||
* Residual race window: the lock and the storage's internal re-resolution
|
||||
* each call this function once. If a PN→LID mapping is added between those
|
||||
* two calls (microsecond window), the lock could be on the PN address while
|
||||
* storage canonicalizes to LID. In production this is mitigated upstream:
|
||||
* `messages-recv.ts` runs `migrateSession()` synchronously BEFORE
|
||||
* `decryptMessage()` is called, so the mapping is already populated by the
|
||||
* time we lock. Eliminating the residual race entirely would require storage
|
||||
* to NOT re-resolve, which is a bigger refactor (other call sites depend on
|
||||
* the current contract).
|
||||
*
|
||||
* Top-level (not inside `signalStorage`) so `makeLibSignalRepository` can
|
||||
* compute the same wire address for transaction locking.
|
||||
*/
|
||||
const resolveLIDSignalAddress = async (id: string, lidMapping: LIDMappingStore): Promise<string> => {
|
||||
if (id.includes('.')) {
|
||||
const [deviceId, device] = id.split('.')
|
||||
if (!deviceId) {
|
||||
throw new Error(`Malformed signal address (empty user portion before '.'): "${id}"`)
|
||||
}
|
||||
|
||||
if (device === undefined || device === '') {
|
||||
throw new Error(`Malformed signal address (empty device portion after '.'): "${id}"`)
|
||||
}
|
||||
|
||||
const [user, domainType_] = deviceId.split('_')
|
||||
const domainType = parseInt(domainType_ || '0')
|
||||
|
||||
if (domainType === WAJIDDomains.LID || domainType === WAJIDDomains.HOSTED_LID) return id
|
||||
|
||||
const pnJid = `${user!}${device !== '0' ? `:${device}` : ''}@${domainType === WAJIDDomains.HOSTED ? 'hosted' : 's.whatsapp.net'}`
|
||||
|
||||
const lidForPN = await lidMapping.getLIDForPN(pnJid)
|
||||
if (lidForPN) {
|
||||
const lidAddr = jidToSignalProtocolAddress(lidForPN)
|
||||
return lidAddr.toString()
|
||||
}
|
||||
}
|
||||
|
||||
return id
|
||||
}
|
||||
|
||||
/**
|
||||
* Extended SignalStorage with identity key management
|
||||
* This type adds identity key operations to the standard Signal storage
|
||||
@@ -815,11 +769,33 @@ function signalStorage(
|
||||
lidMapping: LIDMappingStore,
|
||||
identityKeyCache: LRUCache<string, Uint8Array>,
|
||||
ev?: BaileysEventEmitter,
|
||||
preKeyCircuitBreaker?: CircuitBreaker,
|
||||
logger?: ILogger
|
||||
): ExtendedSignalStorage {
|
||||
// Bind the top-level `resolveLIDSignalAddress` to this instance's lidMapping
|
||||
// so call sites below stay readable.
|
||||
const resolveAddr = (id: string) => resolveLIDSignalAddress(id, lidMapping)
|
||||
// Shared function to resolve PN signal address to LID if mapping exists
|
||||
const resolveLIDSignalAddress = async (id: string): Promise<string> => {
|
||||
if (id.includes('.')) {
|
||||
const [deviceId, device] = id.split('.')
|
||||
if (!deviceId) {
|
||||
throw new Error('Missing device ID')
|
||||
}
|
||||
|
||||
const [user, domainType_] = deviceId.split('_')
|
||||
const domainType = parseInt(domainType_ || '0')
|
||||
|
||||
if (domainType === WAJIDDomains.LID || domainType === WAJIDDomains.HOSTED_LID) return id
|
||||
|
||||
const pnJid = `${user!}${device !== '0' ? `:${device}` : ''}@${domainType === WAJIDDomains.HOSTED ? 'hosted' : 's.whatsapp.net'}`
|
||||
|
||||
const lidForPN = await lidMapping.getLIDForPN(pnJid)
|
||||
if (lidForPN) {
|
||||
const lidAddr = jidToSignalProtocolAddress(lidForPN)
|
||||
return lidAddr.toString()
|
||||
}
|
||||
}
|
||||
|
||||
return id
|
||||
}
|
||||
|
||||
// Delayed PreKey deletion: grace period to handle race conditions
|
||||
// where two pkmsg with the same preKeyId arrive nearly simultaneously.
|
||||
@@ -831,7 +807,7 @@ function signalStorage(
|
||||
return {
|
||||
loadSession: async (id: string) => {
|
||||
try {
|
||||
const wireJid = await resolveAddr(id)
|
||||
const wireJid = await resolveLIDSignalAddress(id)
|
||||
const { [wireJid]: sess } = await keys.get('session', [wireJid])
|
||||
|
||||
if (sess) {
|
||||
@@ -844,7 +820,7 @@ function signalStorage(
|
||||
return null
|
||||
},
|
||||
storeSession: async (id: string, session: libsignal.SessionRecord) => {
|
||||
const wireJid = await resolveAddr(id)
|
||||
const wireJid = await resolveLIDSignalAddress(id)
|
||||
await keys.set({ session: { [wireJid]: session.serialize() } })
|
||||
},
|
||||
isTrustedIdentity: () => {
|
||||
@@ -921,7 +897,7 @@ function signalStorage(
|
||||
const timer = metrics.signalIdentityKeyOperations?.startTimer({ operation: 'load' })
|
||||
|
||||
try {
|
||||
const wireJid = await resolveAddr(id)
|
||||
const wireJid = await resolveLIDSignalAddress(id)
|
||||
|
||||
// Check cache first
|
||||
const cached = identityKeyCache.get(wireJid)
|
||||
@@ -955,7 +931,7 @@ function signalStorage(
|
||||
const timer = metrics.signalIdentityKeyOperations?.startTimer({ operation: 'save' })
|
||||
|
||||
try {
|
||||
const wireJid = await resolveAddr(id)
|
||||
const wireJid = await resolveLIDSignalAddress(id)
|
||||
const currentFingerprint = generateKeyFingerprint(identityKey)
|
||||
|
||||
// Load existing key (from cache or storage)
|
||||
|
||||
+167
-95
@@ -39,7 +39,12 @@ import {
|
||||
xmppSignedPreKey
|
||||
} from '../Utils'
|
||||
import { getPlatformId, isAndroidBrowser } from '../Utils/browser-utils'
|
||||
import { withBoundedRetry } from '../Utils/bounded-retry'
|
||||
import {
|
||||
CircuitBreaker,
|
||||
CircuitOpenError,
|
||||
createConnectionCircuitBreaker,
|
||||
createPreKeyCircuitBreaker
|
||||
} from '../Utils/circuit-breaker'
|
||||
import {
|
||||
decrementActiveConnections,
|
||||
incrementActiveConnections,
|
||||
@@ -89,6 +94,10 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
transactionOpts,
|
||||
qrTimeout,
|
||||
makeSignalRepository,
|
||||
enableCircuitBreaker = true,
|
||||
queryCircuitBreaker: queryCircuitBreakerConfig,
|
||||
connectionCircuitBreaker: connectionCircuitBreakerConfig,
|
||||
preKeyCircuitBreaker: preKeyCircuitBreakerConfig,
|
||||
// If enableUnifiedSession is explicitly set (true/false), use it
|
||||
// Otherwise (undefined), check env var, then default to true
|
||||
enableUnifiedSession: enableUnifiedSessionConfig
|
||||
@@ -98,6 +107,57 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
const enableUnifiedSession =
|
||||
enableUnifiedSessionConfig !== undefined ? enableUnifiedSessionConfig : shouldEnableUnifiedSession()
|
||||
|
||||
// Initialize circuit breakers if enabled
|
||||
let queryCircuitBreaker: CircuitBreaker | undefined
|
||||
let connectionCircuitBreaker: CircuitBreaker | undefined
|
||||
let preKeyCircuitBreaker: CircuitBreaker | undefined
|
||||
|
||||
if (enableCircuitBreaker) {
|
||||
// Circuit breaker for query operations (most critical)
|
||||
queryCircuitBreaker = createConnectionCircuitBreaker({
|
||||
name: 'socket-query',
|
||||
failureThreshold: 5,
|
||||
failureWindow: 60000,
|
||||
resetTimeout: 30000,
|
||||
successThreshold: 2,
|
||||
timeout: defaultQueryTimeoutMs || 60000,
|
||||
onStateChange: (from, to) => {
|
||||
logger.info({ from, to }, 'Query circuit breaker state changed')
|
||||
},
|
||||
onOpen: () => {
|
||||
logger.warn('Query circuit breaker OPENED - blocking requests')
|
||||
},
|
||||
onClose: () => {
|
||||
logger.info('Query circuit breaker CLOSED - resuming normal operation')
|
||||
},
|
||||
...queryCircuitBreakerConfig
|
||||
})
|
||||
|
||||
// Circuit breaker for connection operations
|
||||
connectionCircuitBreaker = createConnectionCircuitBreaker({
|
||||
name: 'socket-connection',
|
||||
failureThreshold: 3,
|
||||
failureWindow: 30000,
|
||||
resetTimeout: 60000,
|
||||
successThreshold: 1,
|
||||
onStateChange: (from, to) => {
|
||||
logger.info({ from, to }, 'Connection circuit breaker state changed')
|
||||
},
|
||||
...connectionCircuitBreakerConfig
|
||||
})
|
||||
|
||||
// Circuit breaker for pre-key operations
|
||||
preKeyCircuitBreaker = createPreKeyCircuitBreaker({
|
||||
name: 'socket-prekey',
|
||||
onStateChange: (from, to) => {
|
||||
logger.info({ from, to }, 'PreKey circuit breaker state changed')
|
||||
},
|
||||
...preKeyCircuitBreakerConfig
|
||||
})
|
||||
|
||||
logger.info('Circuit breakers initialized for socket operations')
|
||||
}
|
||||
|
||||
// Unified Session Manager will be initialized after sendNode is defined
|
||||
let unifiedSessionManager: UnifiedSessionManager | undefined
|
||||
|
||||
@@ -178,17 +238,20 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a raw buffer over the WebSocket.
|
||||
*
|
||||
* Replaces the previous connectionCircuitBreaker wrapping. WebSocket
|
||||
* lifecycle errors (Connection Closed / Lost / Replaced) were already
|
||||
* filtered out of the circuit's failure count, so the only failures it
|
||||
* tripped on were rare native send() errors — for which a state-machine
|
||||
* gate adds no value over a fast propagated error. Reconnection logic
|
||||
* in makeSocket already handles WS recovery independently.
|
||||
*/
|
||||
/** send a raw buffer with circuit breaker protection */
|
||||
const sendRawMessage = async (data: Uint8Array | Buffer) => {
|
||||
if (connectionCircuitBreaker) {
|
||||
try {
|
||||
return await connectionCircuitBreaker.execute(() => sendRawMessageInternal(data))
|
||||
} catch (error) {
|
||||
if (error instanceof CircuitOpenError) {
|
||||
logger.warn({ circuitName: error.circuitName }, 'Send blocked by connection circuit breaker')
|
||||
}
|
||||
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
return sendRawMessageInternal(data)
|
||||
}
|
||||
|
||||
@@ -203,7 +266,7 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
}
|
||||
|
||||
// Initialize Unified Session Manager now that sendNode is defined
|
||||
// (single initialization to avoid duplicating manager state)
|
||||
// (single initialization to avoid duplicating circuit breakers and state)
|
||||
if (enableUnifiedSession) {
|
||||
const sendNodeForSession = async (node: BinaryNode): Promise<void> => {
|
||||
await sendNode(node)
|
||||
@@ -212,6 +275,7 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
unifiedSessionManager = createUnifiedSessionManager({
|
||||
enabled: true,
|
||||
logger,
|
||||
enableCircuitBreaker,
|
||||
sendNode: sendNodeForSession
|
||||
})
|
||||
logger.info('Unified session manager initialized')
|
||||
@@ -282,8 +346,11 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
// Register the response listener BEFORE sending — avoids a race where the server
|
||||
// responds before we start listening. waitForMessage already handles its own
|
||||
// timeout (returns undefined) and connection-close errors (throws), so we do NOT
|
||||
// wrap it in a second promiseTimeout. The outer wrapper would race the inner
|
||||
// timeout near the deadline and throw a spurious Boom('Timed Out').
|
||||
// wrap it in a second promiseTimeout. The outer wrapper caused a race condition
|
||||
// where both timers fired at ~the same deadline: the outer one threw
|
||||
// Boom('Timed Out') while sendNode was still pending, producing a spurious error
|
||||
// whose message contained "socket-query" → matched the circuit-breaker's
|
||||
// "socket" pattern → incorrectly tripped the breaker after 5 timeouts.
|
||||
const responsePromise = waitForMessage<any>(msgId, timeoutMs)
|
||||
// Prevent unhandled-rejection if sendNode throws before we reach
|
||||
// `await responsePromise` below. The error from sendNode still propagates
|
||||
@@ -305,17 +372,23 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a query (iq stanza) and wait for its response.
|
||||
*
|
||||
* Replaces the previous queryCircuitBreaker wrapping (state-machine that
|
||||
* blocked ALL queries for 30s after 5 failures). The cascade behavior
|
||||
* was incompatible with WhatsApp Android's empirical retry pattern:
|
||||
* each operation retries independently with exponential backoff, no
|
||||
* global state. Callers that need retry semantics wrap query() with
|
||||
* withBoundedRetry() at their level (assertSessions, etc.).
|
||||
*/
|
||||
/** send a query with circuit breaker protection */
|
||||
const query = async (node: BinaryNode, timeoutMs?: number) => {
|
||||
// If circuit breaker is enabled, wrap the query
|
||||
if (queryCircuitBreaker) {
|
||||
try {
|
||||
return await queryCircuitBreaker.execute(() => queryInternal(node, timeoutMs))
|
||||
} catch (error) {
|
||||
// If circuit is open, log and rethrow with context
|
||||
if (error instanceof CircuitOpenError) {
|
||||
logger.warn({ circuitName: error.circuitName, state: error.state }, 'Query blocked by circuit breaker')
|
||||
}
|
||||
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback to direct query if circuit breaker is disabled
|
||||
return queryInternal(node, timeoutMs)
|
||||
}
|
||||
|
||||
@@ -612,14 +685,20 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
let lastUploadTime = 0
|
||||
|
||||
/** generates and uploads a set of pre-keys to the server */
|
||||
const uploadPreKeys = async (count = MIN_PREKEY_COUNT) => {
|
||||
// Check minimum interval. (Previously this was guarded by `retryCount === 0`
|
||||
// because the function recursed on retry; with bounded-retry handling
|
||||
// retries internally there is no recursion and the guard is implicit.)
|
||||
const timeSinceLastUpload = Date.now() - lastUploadTime
|
||||
if (timeSinceLastUpload < MIN_UPLOAD_INTERVAL) {
|
||||
logger.debug(`Skipping upload, only ${timeSinceLastUpload}ms since last upload`)
|
||||
return
|
||||
const uploadPreKeys = async (count = MIN_PREKEY_COUNT, retryCount = 0) => {
|
||||
// Check if pre-key circuit breaker is open
|
||||
if (preKeyCircuitBreaker?.isOpen()) {
|
||||
logger.warn('PreKey circuit breaker is open, skipping upload')
|
||||
throw new CircuitOpenError('socket-prekey', 'open')
|
||||
}
|
||||
|
||||
// Check minimum interval (except for retries)
|
||||
if (retryCount === 0) {
|
||||
const timeSinceLastUpload = Date.now() - lastUploadTime
|
||||
if (timeSinceLastUpload < MIN_UPLOAD_INTERVAL) {
|
||||
logger.debug(`Skipping upload, only ${timeSinceLastUpload}ms since last upload`)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Prevent multiple concurrent uploads — if one is already running, wait for it and return:
|
||||
@@ -630,15 +709,8 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
return
|
||||
}
|
||||
|
||||
// Shared abort controller so the outer Promise.race can cancel the
|
||||
// in-flight bounded-retry loop if its own timeout fires first.
|
||||
// Without this, the outer race rejects but the bounded-retry loop
|
||||
// keeps running in the background, mutating lastUploadTime / logs
|
||||
// after the caller has given up (CodeRabbit + Copilot reviews).
|
||||
const uploadAbort = new AbortController()
|
||||
|
||||
const uploadLogic = async () => {
|
||||
logger.info({ count }, 'uploading pre-keys')
|
||||
logger.info({ count, retryCount }, 'uploading pre-keys')
|
||||
|
||||
// Generate and save pre-keys atomically (prevents ID collisions on retry)
|
||||
const node = await keys.transaction(async () => {
|
||||
@@ -649,69 +721,45 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
return node // Only return node since update is already used
|
||||
}, creds?.me?.id || 'upload-pre-keys')
|
||||
|
||||
// Upload to server. Bounded-retry handles backoff (replaces both
|
||||
// the previous circuit breaker AND the manual exponential backoff
|
||||
// retry loop with maxRetries=3).
|
||||
//
|
||||
// CRITICAL: query() returns `undefined` on timeout (it does NOT
|
||||
// throw — see waitForMessage). If we just `await query(node)`,
|
||||
// a timed-out upload would resolve with undefined and bounded-retry
|
||||
// would treat it as success. Validate the response shape so a
|
||||
// timeout actually triggers a retry (CodeRabbit review on PR #393).
|
||||
//
|
||||
// bounded-retry's ttlMs MUST be < UPLOAD_TIMEOUT (the outer
|
||||
// Promise.race below). With margin (28s vs 30s) bounded-retry
|
||||
// always reaches its natural give-up first. We additionally pass
|
||||
// an AbortController signal so that if the outer race ever fires
|
||||
// first, the bounded-retry loop is aborted and does not keep
|
||||
// running in the background (CodeRabbit + Copilot reviews).
|
||||
const PER_ATTEMPT_TIMEOUT_MS = 8_000
|
||||
const RETRY_TTL_MS = UPLOAD_TIMEOUT - 2_000
|
||||
|
||||
const uploadToServer = async (signal?: AbortSignal) => {
|
||||
const result = await query(node, PER_ATTEMPT_TIMEOUT_MS)
|
||||
if (signal?.aborted) {
|
||||
throw new Error('aborted')
|
||||
}
|
||||
|
||||
if (!result) {
|
||||
// query() returned undefined → underlying iq response
|
||||
// timed out. Throw so bounded-retry retries.
|
||||
throw new Boom('Pre-key upload query timed out (no response)', { statusCode: 408 })
|
||||
}
|
||||
|
||||
// Upload to server with circuit breaker protection
|
||||
const uploadToServer = async () => {
|
||||
await query(node)
|
||||
logger.info({ count }, 'uploaded pre-keys successfully')
|
||||
lastUploadTime = Date.now()
|
||||
}
|
||||
|
||||
try {
|
||||
await withBoundedRetry(uploadToServer, {
|
||||
name: 'uploadPreKeys',
|
||||
// 1s -> 2s -> 4s -> 8s (cap). Max 4 attempts fit comfortably
|
||||
// within RETRY_TTL_MS (28s) given perAttemptTimeoutMs=8s.
|
||||
delays: [1000, 2000, 4000, 8000],
|
||||
ttlMs: RETRY_TTL_MS,
|
||||
perAttemptTimeoutMs: PER_ATTEMPT_TIMEOUT_MS,
|
||||
signal: uploadAbort.signal,
|
||||
logger
|
||||
})
|
||||
// Use circuit breaker if available
|
||||
if (preKeyCircuitBreaker) {
|
||||
await preKeyCircuitBreaker.execute(uploadToServer)
|
||||
} else {
|
||||
await uploadToServer()
|
||||
}
|
||||
} catch (uploadError) {
|
||||
logger.error({ uploadError: (uploadError as Error).toString(), count }, 'Failed to upload pre-keys to server')
|
||||
|
||||
// Don't retry if circuit breaker is open
|
||||
if (uploadError instanceof CircuitOpenError) {
|
||||
throw uploadError
|
||||
}
|
||||
|
||||
// Exponential backoff retry (max 3 retries)
|
||||
if (retryCount < 3) {
|
||||
const backoffDelay = Math.min(1000 * Math.pow(2, retryCount), 10000)
|
||||
logger.info(`Retrying pre-key upload in ${backoffDelay}ms`)
|
||||
await new Promise(resolve => setTimeout(resolve, backoffDelay))
|
||||
return uploadPreKeys(count, retryCount + 1)
|
||||
}
|
||||
|
||||
throw uploadError
|
||||
}
|
||||
}
|
||||
|
||||
// Outer timeout protection. With ttlMs=28s vs UPLOAD_TIMEOUT=30s,
|
||||
// bounded-retry should always win — but if anything ever blocks
|
||||
// uploadLogic outside bounded-retry's reach, this race aborts the
|
||||
// whole thing cleanly via uploadAbort.
|
||||
// Add timeout protection
|
||||
uploadPreKeysPromise = Promise.race([
|
||||
uploadLogic(),
|
||||
new Promise<void>((_, reject) =>
|
||||
setTimeout(() => {
|
||||
uploadAbort.abort()
|
||||
reject(new Boom('Pre-key upload timeout', { statusCode: 408 }))
|
||||
}, UPLOAD_TIMEOUT)
|
||||
setTimeout(() => reject(new Boom('Pre-key upload timeout', { statusCode: 408 })), UPLOAD_TIMEOUT)
|
||||
)
|
||||
])
|
||||
|
||||
@@ -1183,6 +1231,12 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
cleanupPreKeyAutoSync()
|
||||
cleanupSessionTTL()
|
||||
|
||||
// CRITICAL: Destroy circuit breakers AFTER cleanup functions complete
|
||||
// This ensures cleanup functions can still use circuit breakers if needed
|
||||
queryCircuitBreaker?.destroy()
|
||||
connectionCircuitBreaker?.destroy()
|
||||
preKeyCircuitBreaker?.destroy()
|
||||
|
||||
// IMPORTANT: Do NOT use removeAllListeners('connection.update')
|
||||
// It would remove consumer listeners, breaking their reconnection logic
|
||||
}
|
||||
@@ -1235,12 +1289,11 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
return // connection closing — do not reschedule
|
||||
} else if (ws.isOpen) {
|
||||
// Send keep-alive ping via sendNode() (fire-and-forget) instead of query().
|
||||
// We don't need the response correlator overhead — WA's pong arrives as an
|
||||
// incoming frame and updates lastDateRecv via the standard receive path.
|
||||
// (Historical context: this used to bypass the query circuit breaker,
|
||||
// which would block pings during cascade failures. Circuit breaker has
|
||||
// since been replaced by per-operation bounded-retry — but the
|
||||
// fire-and-forget pattern stays correct on its own merits.)
|
||||
// query() wraps the ping in the query circuit breaker — when that breaker is
|
||||
// open or timing out, the ping is never sent, WA never responds, lastDateRecv
|
||||
// goes stale, and the diff check above wrongly fires "Connection was lost".
|
||||
// sendNode() bypasses the query circuit breaker entirely; WA's ping response
|
||||
// still arrives as an incoming frame and updates lastDateRecv normally.
|
||||
sendNode({
|
||||
tag: 'iq',
|
||||
attrs: {
|
||||
@@ -1756,6 +1809,25 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
sendWAMBuffer,
|
||||
executeUSyncQuery,
|
||||
onWhatsApp,
|
||||
// Circuit breaker utilities
|
||||
circuitBreakers: {
|
||||
query: queryCircuitBreaker,
|
||||
connection: connectionCircuitBreaker,
|
||||
preKey: preKeyCircuitBreaker
|
||||
},
|
||||
/** Get circuit breaker statistics */
|
||||
getCircuitBreakerStats: () => ({
|
||||
query: queryCircuitBreaker?.getStats(),
|
||||
connection: connectionCircuitBreaker?.getStats(),
|
||||
preKey: preKeyCircuitBreaker?.getStats()
|
||||
}),
|
||||
/** Reset all circuit breakers to closed state */
|
||||
resetCircuitBreakers: () => {
|
||||
queryCircuitBreaker?.reset()
|
||||
connectionCircuitBreaker?.reset()
|
||||
preKeyCircuitBreaker?.reset()
|
||||
logger.info('All circuit breakers reset to closed state')
|
||||
},
|
||||
// Unified Session Telemetry
|
||||
/** Send unified_session telemetry manually */
|
||||
sendUnifiedSession,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import type { Agent } from 'https'
|
||||
import type { URL } from 'url'
|
||||
import { proto } from '../../WAProto/index.js'
|
||||
import type { CircuitBreakerOptions } from '../Utils/circuit-breaker'
|
||||
import type { ILogger } from '../Utils/logger'
|
||||
import type { AuthenticationState, LIDMapping, SignalAuthState, TransactionCapabilityOptions } from './Auth'
|
||||
import type { GroupMetadata } from './GroupMetadata'
|
||||
@@ -202,6 +203,23 @@ export type SocketConfig = {
|
||||
pnToLIDFunc?: (jids: string[]) => Promise<LIDMapping[] | undefined>
|
||||
) => SignalRepositoryWithLIDStore
|
||||
|
||||
// === Circuit Breaker Configuration ===
|
||||
|
||||
/** Enable circuit breaker protection for socket operations (default: true) */
|
||||
enableCircuitBreaker?: boolean
|
||||
|
||||
/** Circuit breaker configuration for query operations */
|
||||
queryCircuitBreaker?: Partial<CircuitBreakerOptions>
|
||||
|
||||
/** Circuit breaker configuration for connection operations */
|
||||
connectionCircuitBreaker?: Partial<CircuitBreakerOptions>
|
||||
|
||||
/** Circuit breaker configuration for pre-key operations */
|
||||
preKeyCircuitBreaker?: Partial<CircuitBreakerOptions>
|
||||
|
||||
/** Circuit breaker configuration for message operations */
|
||||
messageCircuitBreaker?: Partial<CircuitBreakerOptions>
|
||||
|
||||
// === Listener Limits (Memory Leak Prevention) ===
|
||||
|
||||
/**
|
||||
|
||||
@@ -1,393 +0,0 @@
|
||||
/**
|
||||
* Bounded retry — WhatsApp-aligned per-operation retry without global state.
|
||||
*
|
||||
* # Why this exists
|
||||
*
|
||||
* Replaces the circuit breaker pattern that was causing cascading failures
|
||||
* in production: 5 timeouts in 60s would trip the global breaker, blocking
|
||||
* EVERY socket query (typing indicators, profile pic fetches, contact
|
||||
* validation, presence updates) for 30s — even queries to peers that were
|
||||
* perfectly healthy.
|
||||
*
|
||||
* # Empirical justification
|
||||
*
|
||||
* Captured WhatsApp Android's actual retry behavior via Frida hooks
|
||||
* (`hook-circuit-breaker-re-v2.js`, `hook-retry-bounds.js`) on a real
|
||||
* device under controlled WiFi off/on cycles. Findings:
|
||||
*
|
||||
* 1. Delay sequence (per-operation): 3s -> 10s -> 60s -> ~64s -> 120s
|
||||
* (cap at 2 min). Last value reused for further attempts.
|
||||
* 2. Memory profile during 5-min network outage: PSS dropped 53MB and
|
||||
* stabilised. FDs closed (305 -> 295). NO unbounded accumulation.
|
||||
* 3. Recovery on reconnect: ~10s. No retry storm.
|
||||
* 4. NO global state machine observed. Each operation has its own timer.
|
||||
* Failures of operation A do NOT block operation B.
|
||||
*
|
||||
* The default delays in WHATSAPP_BACKOFF_DELAYS below match the captured
|
||||
* sequence directly. The default 10-min TTL is empirical 3-5 min stability
|
||||
* + safety buffer.
|
||||
*
|
||||
* # Design properties
|
||||
*
|
||||
* - Per-operation isolation: each call to `withBoundedRetry` has its own
|
||||
* timer. No shared state. Operation A failing has zero effect on B.
|
||||
* - Bounded by `ttlMs` (wall-clock budget). The TTL is enforced strictly
|
||||
* at three points: (a) before each attempt, (b) cap on per-attempt
|
||||
* timeout, (c) cap on retry delay.
|
||||
* - Per-attempt timeout: each attempt has its own deadline (default 30s),
|
||||
* automatically capped to remaining TTL budget so total runtime never
|
||||
* exceeds ttlMs.
|
||||
* - AbortSignal cancellation: external cancellation supported, both for
|
||||
* the sleep between retries AND (optionally) the in-flight operation
|
||||
* when the operation accepts a signal parameter.
|
||||
* - Memory bound: at most one outstanding retry timer per call. Listeners
|
||||
* are explicitly removed when timers settle. Once the call resolves /
|
||||
* rejects / aborts, all state is freed.
|
||||
*
|
||||
* # Logging
|
||||
*
|
||||
* Pass `logger` in options to get structured logs at each retry attempt,
|
||||
* give-up, and post-failure recovery. The module emits Prometheus metrics
|
||||
* regardless of whether a logger is provided.
|
||||
*
|
||||
* # When to use this vs. plain query()
|
||||
*
|
||||
* - Use plain `query()` when you want fast-fail semantics (caller decides
|
||||
* what to do on failure). Most call sites in InfiniteAPI use this.
|
||||
* - Use `withBoundedRetry(() => query(...), { name: 'X' })` when the
|
||||
* operation is "must-eventually-succeed" with no upstream retry — e.g.
|
||||
* `uploadPreKeys` or other write paths where the alternative is data loss.
|
||||
*
|
||||
* # Examples
|
||||
*
|
||||
* ```ts
|
||||
* // Single attempt with 5-min TTL, give up after that:
|
||||
* await withBoundedRetry(
|
||||
* () => assertSessions([jid], true),
|
||||
* { name: 'assertSessions', ttlMs: 5 * 60_000, logger }
|
||||
* )
|
||||
*
|
||||
* // Tight deadline, fast give-up:
|
||||
* await withBoundedRetry(
|
||||
* () => sendNode(node),
|
||||
* { name: 'send', ttlMs: 30_000, perAttemptTimeoutMs: 5_000 }
|
||||
* )
|
||||
*
|
||||
* // Operation that respects abort signal (best practice):
|
||||
* await withBoundedRetry(
|
||||
* (signal) => fetchSomething({ signal }),
|
||||
* { name: 'fetch', logger }
|
||||
* )
|
||||
* ```
|
||||
*
|
||||
* @module Utils/bounded-retry
|
||||
*/
|
||||
|
||||
import type { ILogger } from './logger'
|
||||
import { metrics } from './prometheus-metrics.js'
|
||||
|
||||
/**
|
||||
* Default delay sequence (milliseconds) — matches WhatsApp Android empirical
|
||||
* behavior. After exhausting the sequence, the last value is used (cap).
|
||||
*/
|
||||
export const WHATSAPP_BACKOFF_DELAYS = [3000, 10000, 60000, 60000, 120000] as const
|
||||
|
||||
/**
|
||||
* Default jitter (+/- 15%) to prevent thundering-herd retries.
|
||||
*/
|
||||
export const DEFAULT_JITTER_FACTOR = 0.15 as const
|
||||
|
||||
/**
|
||||
* Default time-to-live for retries: 10 minutes.
|
||||
*
|
||||
* Empirical justification: WhatsApp Android stabilises memory in ~3-5 min
|
||||
* during a network outage; 10 min gives a generous safety buffer while
|
||||
* preventing unbounded retry accumulation.
|
||||
*/
|
||||
export const DEFAULT_TTL_MS = 10 * 60 * 1000
|
||||
|
||||
/**
|
||||
* Default per-attempt timeout: 30 seconds.
|
||||
*
|
||||
* Most WhatsApp queries respond within seconds. A 30s timeout is generous
|
||||
* enough for slow networks but prevents a single hang from blocking retries.
|
||||
*/
|
||||
export const DEFAULT_PER_ATTEMPT_TIMEOUT_MS = 30000
|
||||
|
||||
export interface BoundedRetryOptions {
|
||||
/** Operation name for logging/metrics */
|
||||
name?: string
|
||||
/** Sequence of delays (ms). Last value is used as cap. */
|
||||
delays?: readonly number[]
|
||||
/** Jitter factor 0..1 (default 0.15) */
|
||||
jitter?: number
|
||||
/** Total wall-clock budget — gives up after this. Default 10 min. */
|
||||
ttlMs?: number
|
||||
/** Per-attempt timeout (ms). Default 30s. Capped by remaining TTL. */
|
||||
perAttemptTimeoutMs?: number
|
||||
/** Predicate: should we retry on this error? Default: always */
|
||||
shouldRetry?: (err: Error, attempt: number) => boolean
|
||||
/** Hook fired before each retry */
|
||||
onRetry?: (err: Error, attempt: number, delayMs: number) => void
|
||||
/**
|
||||
* AbortSignal — when fired, the loop stops at the next observation point:
|
||||
* - If the loop is sleeping between retries, the sleep rejects immediately.
|
||||
* - If an attempt is in flight, the abort is forwarded to the operation
|
||||
* via the per-attempt signal. The operation must ITSELF observe the
|
||||
* signal (e.g. `(signal) => fetch({ signal })`) for cancellation to
|
||||
* be truly immediate; otherwise it cancels at the next per-attempt
|
||||
* timeout boundary.
|
||||
* - Either way, on the next loop iteration BoundedRetryAbortedError is
|
||||
* thrown, so the caller never sees more than one trailing attempt
|
||||
* after the abort.
|
||||
*/
|
||||
signal?: AbortSignal
|
||||
/** Optional logger for structured retry/give-up/recovery logs */
|
||||
logger?: ILogger
|
||||
}
|
||||
|
||||
export class BoundedRetryGiveUpError extends Error {
|
||||
constructor(
|
||||
public readonly opName: string,
|
||||
public readonly attempts: number,
|
||||
public readonly elapsedMs: number,
|
||||
public readonly lastError: Error
|
||||
) {
|
||||
super(
|
||||
`bounded-retry "${opName}" gave up after ${attempts} attempts ` +
|
||||
`(${elapsedMs}ms elapsed). Last error: ${lastError.message}`
|
||||
)
|
||||
this.name = 'BoundedRetryGiveUpError'
|
||||
}
|
||||
}
|
||||
|
||||
export class BoundedRetryAbortedError extends Error {
|
||||
constructor(public readonly opName: string) {
|
||||
super(`bounded-retry "${opName}" aborted via signal`)
|
||||
this.name = 'BoundedRetryAbortedError'
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply jitter to a delay: returns delay * (1 +/- jitter)
|
||||
*/
|
||||
function withJitter(delayMs: number, jitter: number): number {
|
||||
if (jitter <= 0) return delayMs
|
||||
const factor = 1 + (Math.random() * 2 - 1) * jitter
|
||||
return Math.max(0, Math.round(delayMs * factor))
|
||||
}
|
||||
|
||||
/**
|
||||
* Pick the delay for a given attempt index. Falls back to the last value
|
||||
* (cap) once the sequence is exhausted.
|
||||
*/
|
||||
function pickDelay(attempt: number, delays: readonly number[]): number {
|
||||
if (delays.length === 0) return 0
|
||||
if (attempt < delays.length) return delays[attempt]!
|
||||
return delays[delays.length - 1]!
|
||||
}
|
||||
|
||||
/**
|
||||
* Wrap a promise with a per-attempt timeout. Aborts the supplied controller
|
||||
* when the timeout fires so the caller can cancel any in-flight work
|
||||
* (operations that accept the signal). Always clears the timer on settlement.
|
||||
*/
|
||||
function withTimeout<T>(
|
||||
promise: Promise<T>,
|
||||
timeoutMs: number,
|
||||
name: string,
|
||||
abortOnTimeout: AbortController
|
||||
): Promise<T> {
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
abortOnTimeout.abort()
|
||||
reject(new Error(`bounded-retry "${name}" attempt timed out after ${timeoutMs}ms`))
|
||||
}, timeoutMs)
|
||||
promise
|
||||
.then(value => {
|
||||
clearTimeout(timer)
|
||||
resolve(value)
|
||||
})
|
||||
.catch(err => {
|
||||
clearTimeout(timer)
|
||||
reject(err as Error)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Sleep with abort support. Always removes the abort listener on settlement
|
||||
* so listeners do not accumulate on long-lived signals.
|
||||
*/
|
||||
function sleep(ms: number, signal?: AbortSignal): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
if (signal?.aborted) {
|
||||
reject(new Error('aborted'))
|
||||
return
|
||||
}
|
||||
|
||||
let onAbort: (() => void) | undefined
|
||||
const cleanup = () => {
|
||||
if (onAbort && signal) {
|
||||
signal.removeEventListener('abort', onAbort)
|
||||
}
|
||||
}
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
cleanup()
|
||||
resolve()
|
||||
}, ms)
|
||||
|
||||
if (signal) {
|
||||
onAbort = () => {
|
||||
clearTimeout(timer)
|
||||
cleanup()
|
||||
reject(new Error('aborted'))
|
||||
}
|
||||
|
||||
signal.addEventListener('abort', onAbort, { once: true })
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Run an async operation with bounded exponential-backoff retry.
|
||||
*
|
||||
* Independent per-call: no global state, no cross-operation interaction.
|
||||
* Memory bound: at most one outstanding retry timer per call.
|
||||
*
|
||||
* The operation may optionally accept an `AbortSignal` parameter — when the
|
||||
* per-attempt timeout fires (or the outer signal is aborted), the inner
|
||||
* signal is aborted so the operation can stop in-flight work cleanly.
|
||||
*
|
||||
* @example
|
||||
* ```ts
|
||||
* const result = await withBoundedRetry(
|
||||
* (signal) => assertSessions([jid], true, { signal }),
|
||||
* { name: 'assertSessions', ttlMs: 5 * 60_000, logger }
|
||||
* )
|
||||
* ```
|
||||
*/
|
||||
export async function withBoundedRetry<T>(
|
||||
operation: (signal?: AbortSignal) => Promise<T>,
|
||||
options: BoundedRetryOptions = {}
|
||||
): Promise<T> {
|
||||
const name = options.name ?? 'bounded-retry'
|
||||
const delays = options.delays ?? WHATSAPP_BACKOFF_DELAYS
|
||||
const jitter = options.jitter ?? DEFAULT_JITTER_FACTOR
|
||||
const ttlMs = options.ttlMs ?? DEFAULT_TTL_MS
|
||||
const perAttemptTimeoutMs = options.perAttemptTimeoutMs ?? DEFAULT_PER_ATTEMPT_TIMEOUT_MS
|
||||
const shouldRetry = options.shouldRetry ?? (() => true)
|
||||
const logger = options.logger
|
||||
|
||||
const start = Date.now()
|
||||
let lastError: Error = new Error('unknown')
|
||||
let attempt = 0
|
||||
|
||||
while (true) {
|
||||
// Check abort + TTL BEFORE starting an attempt — strict wall-clock
|
||||
// budget. Without this check the loop could begin a new attempt with
|
||||
// 0ms remaining and then run for up to `perAttemptTimeoutMs`.
|
||||
if (options.signal?.aborted) {
|
||||
throw new BoundedRetryAbortedError(name)
|
||||
}
|
||||
|
||||
const elapsedBeforeAttempt = Date.now() - start
|
||||
const remainingBudget = ttlMs - elapsedBeforeAttempt
|
||||
if (remainingBudget <= 0) {
|
||||
metrics.errors?.inc({ category: 'bounded_retry', code: 'ttl_exceeded' })
|
||||
logger?.warn?.(
|
||||
{ op: name, attempts: attempt, elapsedMs: elapsedBeforeAttempt, ttlMs, lastError: lastError.message },
|
||||
'bounded-retry: TTL exceeded before next attempt — giving up'
|
||||
)
|
||||
throw new BoundedRetryGiveUpError(name, attempt, elapsedBeforeAttempt, lastError)
|
||||
}
|
||||
|
||||
// Cap the per-attempt timeout to the remaining TTL budget so a single
|
||||
// attempt cannot run past the wall-clock deadline.
|
||||
const attemptTimeoutMs = Math.min(perAttemptTimeoutMs, remainingBudget)
|
||||
const attemptAbort = new AbortController()
|
||||
|
||||
// Forward outer abort to the per-attempt controller so the in-flight
|
||||
// operation is cancelled when the user aborts. Cleaned up below.
|
||||
let onOuterAbort: (() => void) | undefined
|
||||
if (options.signal) {
|
||||
onOuterAbort = () => attemptAbort.abort()
|
||||
if (options.signal.aborted) {
|
||||
attemptAbort.abort()
|
||||
} else {
|
||||
options.signal.addEventListener('abort', onOuterAbort, { once: true })
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await withTimeout(operation(attemptAbort.signal), attemptTimeoutMs, name, attemptAbort)
|
||||
|
||||
if (attempt > 0) {
|
||||
metrics.socketEvents?.inc({ event: 'bounded_retry_recovered' })
|
||||
logger?.info?.(
|
||||
{ op: name, attempts: attempt + 1, elapsedMs: Date.now() - start },
|
||||
'bounded-retry: operation succeeded after retries'
|
||||
)
|
||||
}
|
||||
|
||||
return result
|
||||
} catch (err) {
|
||||
lastError = err as Error
|
||||
attempt++
|
||||
|
||||
// (outer-abort listener detachment happens in the finally below —
|
||||
// avoiding a double-remove that breaks listener-count assertions.)
|
||||
|
||||
// If the outer signal aborted us mid-attempt, surface that explicitly
|
||||
// rather than as a generic operation failure.
|
||||
if (options.signal?.aborted) {
|
||||
throw new BoundedRetryAbortedError(name)
|
||||
}
|
||||
|
||||
const elapsed = Date.now() - start
|
||||
|
||||
if (!shouldRetry(lastError, attempt)) {
|
||||
metrics.errors?.inc({ category: 'bounded_retry', code: 'predicate_no_retry' })
|
||||
logger?.warn?.(
|
||||
{ op: name, attempts: attempt, elapsedMs: elapsed, error: lastError.message },
|
||||
'bounded-retry: shouldRetry returned false — giving up'
|
||||
)
|
||||
throw lastError
|
||||
}
|
||||
|
||||
// Re-check budget after the failure so we do not sleep past TTL.
|
||||
const remainingAfterFailure = ttlMs - elapsed
|
||||
if (remainingAfterFailure <= 0) {
|
||||
metrics.errors?.inc({ category: 'bounded_retry', code: 'ttl_exceeded' })
|
||||
logger?.warn?.(
|
||||
{ op: name, attempts: attempt, elapsedMs: elapsed, ttlMs, lastError: lastError.message },
|
||||
'bounded-retry: TTL exceeded after attempt — giving up'
|
||||
)
|
||||
throw new BoundedRetryGiveUpError(name, attempt, elapsed, lastError)
|
||||
}
|
||||
|
||||
const baseDelay = pickDelay(attempt - 1, delays)
|
||||
const delayMs = Math.min(withJitter(baseDelay, jitter), remainingAfterFailure)
|
||||
|
||||
options.onRetry?.(lastError, attempt, delayMs)
|
||||
metrics.socketEvents?.inc({ event: 'bounded_retry_attempt' })
|
||||
logger?.debug?.(
|
||||
{ op: name, attempt, delayMs, elapsedMs: elapsed, error: lastError.message },
|
||||
'bounded-retry: scheduling next attempt'
|
||||
)
|
||||
|
||||
try {
|
||||
await sleep(delayMs, options.signal)
|
||||
} catch {
|
||||
throw new BoundedRetryAbortedError(name)
|
||||
}
|
||||
} finally {
|
||||
// Belt-and-suspenders: ensure the outer-abort listener is always
|
||||
// removed even on early throws.
|
||||
if (options.signal && onOuterAbort) {
|
||||
options.signal.removeEventListener('abort', onOuterAbort)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,766 @@
|
||||
/**
|
||||
* Circuit Breaker Pattern Implementation
|
||||
*
|
||||
* Provides protection against cascading failures by monitoring operation outcomes
|
||||
* and temporarily blocking requests when failure thresholds are exceeded.
|
||||
*
|
||||
* States:
|
||||
* - CLOSED: Normal operation, requests pass through
|
||||
* - OPEN: Failures exceeded threshold, requests are blocked
|
||||
* - HALF_OPEN: Testing recovery, limited requests allowed
|
||||
*
|
||||
* @module Utils/circuit-breaker
|
||||
*/
|
||||
|
||||
import { EventEmitter } from 'events'
|
||||
import { metrics } from './prometheus-metrics.js'
|
||||
|
||||
/**
|
||||
* Circuit breaker operational states
|
||||
*/
|
||||
export type CircuitState = 'closed' | 'open' | 'half-open'
|
||||
|
||||
/**
|
||||
* Failure record with timestamp for sliding window tracking
|
||||
*/
|
||||
export interface FailureRecord {
|
||||
timestamp: number
|
||||
error: Error
|
||||
}
|
||||
|
||||
/**
|
||||
* Circuit breaker configuration options
|
||||
*/
|
||||
export interface CircuitBreakerOptions {
|
||||
/** Unique identifier for this circuit breaker (used in metrics/logging) */
|
||||
name: string
|
||||
/** Number of failures within the window to trigger OPEN state (default: 5) */
|
||||
failureThreshold?: number
|
||||
/** Time window in ms for counting failures (default: 60000) */
|
||||
failureWindow?: number
|
||||
/** Number of successes required in HALF_OPEN to return to CLOSED (default: 2) */
|
||||
successThreshold?: number
|
||||
/** Time in ms to wait before transitioning from OPEN to HALF_OPEN (default: 30000) */
|
||||
resetTimeout?: number
|
||||
/** Timeout for individual operations in ms (default: 10000) */
|
||||
timeout?: number
|
||||
/** Minimum number of requests before circuit can trip (default: 5) */
|
||||
volumeThreshold?: number
|
||||
/** Predicate to determine if an error should count as a failure */
|
||||
shouldCountError?: (error: Error) => boolean
|
||||
/** Whether to collect Prometheus metrics (default: true) */
|
||||
collectMetrics?: boolean
|
||||
/** Fallback function when circuit is OPEN */
|
||||
fallback?: <T>() => T | Promise<T>
|
||||
/** Callback when state changes */
|
||||
onStateChange?: (from: CircuitState, to: CircuitState) => void
|
||||
/** Callback on failure */
|
||||
onFailure?: (error: Error) => void
|
||||
/** Callback on success */
|
||||
onSuccess?: () => void
|
||||
/** Callback when circuit opens */
|
||||
onOpen?: () => void
|
||||
/** Callback when circuit closes */
|
||||
onClose?: () => void
|
||||
/** Callback when circuit enters half-open */
|
||||
onHalfOpen?: () => void
|
||||
}
|
||||
|
||||
/**
|
||||
* Circuit breaker statistics
|
||||
*/
|
||||
export interface CircuitBreakerStats {
|
||||
state: CircuitState
|
||||
failures: number
|
||||
successes: number
|
||||
consecutiveFailures: number
|
||||
consecutiveSuccesses: number
|
||||
totalCalls: number
|
||||
totalFailures: number
|
||||
totalSuccesses: number
|
||||
totalRejected: number
|
||||
failureRate: number
|
||||
lastFailureTime?: number
|
||||
lastSuccessTime?: number
|
||||
lastStateChange?: number
|
||||
isOpen: boolean
|
||||
isClosed: boolean
|
||||
isHalfOpen: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Error thrown when circuit is OPEN and request is rejected
|
||||
*/
|
||||
export class CircuitOpenError extends Error {
|
||||
constructor(
|
||||
public readonly circuitName: string,
|
||||
public readonly state: CircuitState
|
||||
) {
|
||||
super(`Circuit breaker "${circuitName}" is ${state}`)
|
||||
this.name = 'CircuitOpenError'
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Error thrown when operation exceeds timeout
|
||||
*/
|
||||
export class CircuitTimeoutError extends Error {
|
||||
constructor(
|
||||
public readonly circuitName: string,
|
||||
public readonly timeoutMs: number
|
||||
) {
|
||||
super(`Circuit breaker "${circuitName}" operation timed out after ${timeoutMs}ms`)
|
||||
this.name = 'CircuitTimeoutError'
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Circuit Breaker implementation with sliding window failure tracking
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const breaker = new CircuitBreaker({
|
||||
* name: 'whatsapp-api',
|
||||
* failureThreshold: 5,
|
||||
* resetTimeout: 30000
|
||||
* })
|
||||
*
|
||||
* try {
|
||||
* const result = await breaker.execute(() => sendMessage(msg))
|
||||
* } catch (error) {
|
||||
* if (error instanceof CircuitOpenError) {
|
||||
* // Circuit is open, use fallback
|
||||
* }
|
||||
* }
|
||||
* ```
|
||||
*/
|
||||
export class CircuitBreaker extends EventEmitter {
|
||||
private state: CircuitState = 'closed'
|
||||
private failureRecords: FailureRecord[] = []
|
||||
private consecutiveFailures = 0
|
||||
private consecutiveSuccesses = 0
|
||||
private totalCalls = 0
|
||||
private totalFailures = 0
|
||||
private totalSuccesses = 0
|
||||
private totalRejected = 0
|
||||
private lastFailureTime?: number
|
||||
private lastSuccessTime?: number
|
||||
private lastStateChange: number
|
||||
private resetTimer?: ReturnType<typeof setTimeout>
|
||||
private readonly options: Required<CircuitBreakerOptions>
|
||||
|
||||
constructor(options: CircuitBreakerOptions) {
|
||||
super()
|
||||
|
||||
this.options = {
|
||||
name: options.name,
|
||||
failureThreshold: options.failureThreshold ?? 5,
|
||||
failureWindow: options.failureWindow ?? 60000,
|
||||
successThreshold: options.successThreshold ?? 2,
|
||||
resetTimeout: options.resetTimeout ?? 30000,
|
||||
timeout: options.timeout ?? 10000,
|
||||
volumeThreshold: options.volumeThreshold ?? 5,
|
||||
shouldCountError: options.shouldCountError ?? (() => true),
|
||||
collectMetrics: options.collectMetrics ?? true,
|
||||
fallback:
|
||||
options.fallback ??
|
||||
(() => {
|
||||
throw new CircuitOpenError(this.options.name, this.state)
|
||||
}),
|
||||
onStateChange: options.onStateChange ?? (() => {}),
|
||||
onFailure: options.onFailure ?? (() => {}),
|
||||
onSuccess: options.onSuccess ?? (() => {}),
|
||||
onOpen: options.onOpen ?? (() => {}),
|
||||
onClose: options.onClose ?? (() => {}),
|
||||
onHalfOpen: options.onHalfOpen ?? (() => {})
|
||||
}
|
||||
|
||||
this.lastStateChange = Date.now()
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if the circuit allows execution
|
||||
*/
|
||||
canExecute(): boolean {
|
||||
if (this.state === 'closed') {
|
||||
return true
|
||||
}
|
||||
|
||||
if (this.state === 'open') {
|
||||
return false
|
||||
}
|
||||
|
||||
// HALF_OPEN: allow limited requests for testing
|
||||
return true
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute an async operation with circuit breaker protection
|
||||
*/
|
||||
async execute<T>(operation: () => T | Promise<T>): Promise<T> {
|
||||
this.totalCalls++
|
||||
|
||||
// Check if circuit allows execution
|
||||
if (this.state === 'open') {
|
||||
this.totalRejected++
|
||||
|
||||
if (this.options.collectMetrics) {
|
||||
metrics.errors.inc({ category: 'circuit_breaker', code: 'rejected' })
|
||||
}
|
||||
|
||||
return this.options.fallback() as T
|
||||
}
|
||||
|
||||
// Execute with timeout protection
|
||||
try {
|
||||
const result = await this.executeWithTimeout(operation)
|
||||
this.recordSuccess()
|
||||
return result
|
||||
} catch (error) {
|
||||
this.recordFailure(error as Error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a synchronous operation with circuit breaker protection
|
||||
*/
|
||||
executeSync<T>(operation: () => T): T {
|
||||
this.totalCalls++
|
||||
|
||||
if (this.state === 'open') {
|
||||
this.totalRejected++
|
||||
|
||||
if (this.options.collectMetrics) {
|
||||
metrics.errors.inc({ category: 'circuit_breaker', code: 'rejected' })
|
||||
}
|
||||
|
||||
return this.options.fallback() as T
|
||||
}
|
||||
|
||||
try {
|
||||
const result = operation()
|
||||
this.recordSuccess()
|
||||
return result
|
||||
} catch (error) {
|
||||
this.recordFailure(error as Error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute operation with timeout wrapper
|
||||
*/
|
||||
private async executeWithTimeout<T>(operation: () => T | Promise<T>): Promise<T> {
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
reject(new CircuitTimeoutError(this.options.name, this.options.timeout))
|
||||
}, this.options.timeout)
|
||||
|
||||
Promise.resolve(operation())
|
||||
.then(result => {
|
||||
clearTimeout(timer)
|
||||
resolve(result)
|
||||
})
|
||||
.catch(error => {
|
||||
clearTimeout(timer)
|
||||
reject(error)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Record a successful operation
|
||||
*/
|
||||
private recordSuccess(): void {
|
||||
this.totalSuccesses++
|
||||
this.lastSuccessTime = Date.now()
|
||||
this.consecutiveSuccesses++
|
||||
this.consecutiveFailures = 0
|
||||
|
||||
if (this.options.collectMetrics) {
|
||||
metrics.socketEvents.inc({ event: 'circuit_success' })
|
||||
}
|
||||
|
||||
this.options.onSuccess()
|
||||
this.emit('success')
|
||||
|
||||
// In HALF_OPEN state, check if we can close the circuit
|
||||
if (this.state === 'half-open') {
|
||||
if (this.consecutiveSuccesses >= this.options.successThreshold) {
|
||||
this.transitionTo('closed')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Record a failed operation
|
||||
*/
|
||||
private recordFailure(error: Error): void {
|
||||
// Check if this error should count as a failure
|
||||
if (!this.options.shouldCountError(error)) {
|
||||
return
|
||||
}
|
||||
|
||||
const now = Date.now()
|
||||
this.totalFailures++
|
||||
this.lastFailureTime = now
|
||||
this.consecutiveFailures++
|
||||
this.consecutiveSuccesses = 0
|
||||
|
||||
// Add to failure records for sliding window
|
||||
this.failureRecords.push({ timestamp: now, error })
|
||||
|
||||
// Clean old failures outside the window
|
||||
this.cleanOldFailures()
|
||||
|
||||
if (this.options.collectMetrics) {
|
||||
metrics.errors.inc({ category: 'circuit_breaker', code: 'failure' })
|
||||
}
|
||||
|
||||
this.options.onFailure(error)
|
||||
this.emit('failure', error)
|
||||
|
||||
// State transition logic
|
||||
if (this.state === 'half-open') {
|
||||
// Any failure in HALF_OPEN immediately reopens the circuit
|
||||
this.transitionTo('open')
|
||||
} else if (this.state === 'closed') {
|
||||
// Check if we should trip the circuit
|
||||
const recentFailures = this.failureRecords.length
|
||||
|
||||
if (this.totalCalls >= this.options.volumeThreshold && recentFailures >= this.options.failureThreshold) {
|
||||
this.transitionTo('open')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove failure records outside the sliding window
|
||||
*/
|
||||
private cleanOldFailures(): void {
|
||||
const cutoff = Date.now() - this.options.failureWindow
|
||||
this.failureRecords = this.failureRecords.filter(record => record.timestamp > cutoff)
|
||||
}
|
||||
|
||||
/**
|
||||
* Transition to a new state
|
||||
*/
|
||||
private transitionTo(newState: CircuitState): void {
|
||||
const oldState = this.state
|
||||
|
||||
if (oldState === newState) {
|
||||
return
|
||||
}
|
||||
|
||||
this.state = newState
|
||||
this.lastStateChange = Date.now()
|
||||
|
||||
// Clear existing reset timer
|
||||
if (this.resetTimer) {
|
||||
clearTimeout(this.resetTimer)
|
||||
this.resetTimer = undefined
|
||||
}
|
||||
|
||||
// State-specific actions
|
||||
switch (newState) {
|
||||
case 'closed':
|
||||
this.consecutiveFailures = 0
|
||||
this.consecutiveSuccesses = 0
|
||||
this.failureRecords = []
|
||||
this.options.onClose()
|
||||
this.emit('close')
|
||||
break
|
||||
|
||||
case 'open':
|
||||
this.consecutiveSuccesses = 0
|
||||
this.options.onOpen()
|
||||
this.emit('open')
|
||||
|
||||
// Record circuit breaker trip metric
|
||||
if (this.options.collectMetrics) {
|
||||
metrics.circuitBreakerTrips?.inc({ name: this.options.name })
|
||||
}
|
||||
|
||||
// Schedule transition to HALF_OPEN
|
||||
this.resetTimer = setTimeout(() => {
|
||||
this.transitionTo('half-open')
|
||||
}, this.options.resetTimeout)
|
||||
break
|
||||
|
||||
case 'half-open':
|
||||
this.consecutiveSuccesses = 0
|
||||
this.consecutiveFailures = 0
|
||||
this.options.onHalfOpen()
|
||||
this.emit('half-open')
|
||||
break
|
||||
}
|
||||
|
||||
this.options.onStateChange(oldState, newState)
|
||||
this.emit('state-change', { from: oldState, to: newState })
|
||||
}
|
||||
|
||||
/**
|
||||
* Manually trip the circuit to OPEN state
|
||||
*/
|
||||
trip(): void {
|
||||
this.transitionTo('open')
|
||||
}
|
||||
|
||||
/**
|
||||
* Manually reset the circuit to CLOSED state
|
||||
*/
|
||||
reset(): void {
|
||||
this.transitionTo('closed')
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current circuit state
|
||||
*/
|
||||
getState(): CircuitState {
|
||||
return this.state
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if circuit is OPEN
|
||||
*/
|
||||
isOpen(): boolean {
|
||||
return this.state === 'open'
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if circuit is CLOSED
|
||||
*/
|
||||
isClosed(): boolean {
|
||||
return this.state === 'closed'
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if circuit is HALF_OPEN
|
||||
*/
|
||||
isHalfOpen(): boolean {
|
||||
return this.state === 'half-open'
|
||||
}
|
||||
|
||||
/**
|
||||
* Get circuit breaker statistics
|
||||
*/
|
||||
getStats(): CircuitBreakerStats {
|
||||
this.cleanOldFailures()
|
||||
|
||||
const failureRate = this.totalCalls > 0 ? (this.totalFailures / this.totalCalls) * 100 : 0
|
||||
|
||||
return {
|
||||
state: this.state,
|
||||
failures: this.failureRecords.length,
|
||||
successes: this.consecutiveSuccesses,
|
||||
consecutiveFailures: this.consecutiveFailures,
|
||||
consecutiveSuccesses: this.consecutiveSuccesses,
|
||||
totalCalls: this.totalCalls,
|
||||
totalFailures: this.totalFailures,
|
||||
totalSuccesses: this.totalSuccesses,
|
||||
totalRejected: this.totalRejected,
|
||||
failureRate,
|
||||
lastFailureTime: this.lastFailureTime,
|
||||
lastSuccessTime: this.lastSuccessTime,
|
||||
lastStateChange: this.lastStateChange,
|
||||
isOpen: this.isOpen(),
|
||||
isClosed: this.isClosed(),
|
||||
isHalfOpen: this.isHalfOpen()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get circuit breaker name
|
||||
*/
|
||||
getName(): string {
|
||||
return this.options.name
|
||||
}
|
||||
|
||||
/**
|
||||
* Destroy circuit breaker and clean up resources
|
||||
*/
|
||||
destroy(): void {
|
||||
if (this.resetTimer) {
|
||||
clearTimeout(this.resetTimer)
|
||||
}
|
||||
|
||||
this.failureRecords = []
|
||||
this.removeAllListeners()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Factory function to create a circuit breaker
|
||||
*/
|
||||
export function createCircuitBreaker(options: CircuitBreakerOptions): CircuitBreaker {
|
||||
return new CircuitBreaker(options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Registry for managing multiple circuit breakers
|
||||
*/
|
||||
export class CircuitBreakerRegistry {
|
||||
private breakers: Map<string, CircuitBreaker> = new Map()
|
||||
|
||||
/**
|
||||
* Get or create a circuit breaker by name
|
||||
*/
|
||||
get(name: string, options?: Omit<CircuitBreakerOptions, 'name'>): CircuitBreaker {
|
||||
if (!this.breakers.has(name)) {
|
||||
const breaker = new CircuitBreaker({ ...options, name })
|
||||
this.breakers.set(name, breaker)
|
||||
}
|
||||
|
||||
return this.breakers.get(name)!
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a circuit breaker exists
|
||||
*/
|
||||
has(name: string): boolean {
|
||||
return this.breakers.has(name)
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove a circuit breaker
|
||||
*/
|
||||
remove(name: string): boolean {
|
||||
const breaker = this.breakers.get(name)
|
||||
if (breaker) {
|
||||
breaker.destroy()
|
||||
return this.breakers.delete(name)
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all circuit breakers
|
||||
*/
|
||||
getAll(): Map<string, CircuitBreaker> {
|
||||
return new Map(this.breakers)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get statistics for all circuit breakers
|
||||
*/
|
||||
getAllStats(): Record<string, CircuitBreakerStats> {
|
||||
const stats: Record<string, CircuitBreakerStats> = {}
|
||||
for (const [name, breaker] of this.breakers) {
|
||||
stats[name] = breaker.getStats()
|
||||
}
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
/**
|
||||
* Reset all circuit breakers to CLOSED state
|
||||
*/
|
||||
resetAll(): void {
|
||||
for (const breaker of this.breakers.values()) {
|
||||
breaker.reset()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Destroy all circuit breakers
|
||||
*/
|
||||
destroyAll(): void {
|
||||
for (const breaker of this.breakers.values()) {
|
||||
breaker.destroy()
|
||||
}
|
||||
|
||||
this.breakers.clear()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Global circuit breaker registry instance
|
||||
*/
|
||||
export const globalCircuitRegistry = new CircuitBreakerRegistry()
|
||||
|
||||
/**
|
||||
* Decorator to protect a method with circuit breaker
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* class MyService {
|
||||
* @circuitBreaker({ failureThreshold: 3 })
|
||||
* async fetchData() {
|
||||
* return await api.getData()
|
||||
* }
|
||||
* }
|
||||
* ```
|
||||
*/
|
||||
export function circuitBreaker(options: Omit<CircuitBreakerOptions, 'name'> & { name?: string } = {}) {
|
||||
return function (
|
||||
_target: unknown,
|
||||
propertyKey: string,
|
||||
descriptor: TypedPropertyDescriptor<(...args: unknown[]) => unknown>
|
||||
) {
|
||||
const originalMethod = descriptor.value
|
||||
if (!originalMethod) return descriptor
|
||||
|
||||
const name = options.name || propertyKey
|
||||
const breaker = globalCircuitRegistry.get(name, options)
|
||||
|
||||
descriptor.value = async function (...args: unknown[]): Promise<unknown> {
|
||||
return breaker.execute(() => originalMethod.apply(this, args))
|
||||
}
|
||||
|
||||
return descriptor
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Wrap a function with circuit breaker protection
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const protectedFetch = withCircuitBreaker(
|
||||
* fetchData,
|
||||
* { name: 'api-fetch', failureThreshold: 5 }
|
||||
* )
|
||||
* ```
|
||||
*/
|
||||
// eslint-disable-next-line space-before-function-paren
|
||||
export function withCircuitBreaker<T extends (...args: unknown[]) => unknown>(
|
||||
fn: T,
|
||||
options: CircuitBreakerOptions
|
||||
): T {
|
||||
const breaker = new CircuitBreaker(options)
|
||||
|
||||
return (async (...args: Parameters<T>): Promise<ReturnType<T>> => {
|
||||
return breaker.execute(() => fn(...args)) as Promise<ReturnType<T>>
|
||||
}) as unknown as T
|
||||
}
|
||||
|
||||
/**
|
||||
* Get health status of all circuit breakers
|
||||
*/
|
||||
export function getCircuitHealth(): {
|
||||
healthy: boolean
|
||||
openCircuits: string[]
|
||||
stats: Record<string, CircuitBreakerStats>
|
||||
} {
|
||||
const stats = globalCircuitRegistry.getAllStats()
|
||||
const openCircuits: string[] = []
|
||||
|
||||
for (const [name, stat] of Object.entries(stats)) {
|
||||
if (stat.isOpen) {
|
||||
openCircuits.push(name)
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
healthy: openCircuits.length === 0,
|
||||
openCircuits,
|
||||
stats
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a pre-configured circuit breaker for WhatsApp PreKey operations
|
||||
*
|
||||
* This circuit breaker is optimized for handling encryption/session errors
|
||||
* that commonly occur with WhatsApp's Signal protocol implementation.
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const preKeyBreaker = createPreKeyCircuitBreaker()
|
||||
*
|
||||
* async function sendEncryptedMessage(msg) {
|
||||
* return preKeyBreaker.execute(async () => {
|
||||
* return await encryptAndSend(msg)
|
||||
* })
|
||||
* }
|
||||
* ```
|
||||
*/
|
||||
export function createPreKeyCircuitBreaker(customOptions?: Partial<CircuitBreakerOptions>): CircuitBreaker {
|
||||
const preKeyErrorPatterns = ['prekey', 'pre-key', 'session', 'signal', 'encrypt', 'decrypt', 'cipher', 'key']
|
||||
|
||||
return new CircuitBreaker({
|
||||
name: 'prekey-operations',
|
||||
failureThreshold: 5,
|
||||
failureWindow: 60000,
|
||||
resetTimeout: 30000,
|
||||
successThreshold: 2,
|
||||
shouldCountError: (error: Error) => {
|
||||
const message = error.message.toLowerCase()
|
||||
return preKeyErrorPatterns.some(pattern => message.includes(pattern))
|
||||
},
|
||||
...customOptions
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a circuit breaker for WebSocket connection operations
|
||||
*/
|
||||
export function createConnectionCircuitBreaker(customOptions?: Partial<CircuitBreakerOptions>): CircuitBreaker {
|
||||
const connectionErrorPatterns = [
|
||||
'econnrefused',
|
||||
'econnreset',
|
||||
'etimedout',
|
||||
'enotfound',
|
||||
'socket',
|
||||
'websocket',
|
||||
'connection',
|
||||
'network'
|
||||
]
|
||||
|
||||
// Normal WS lifecycle status codes that must NEVER trip the query circuit breaker.
|
||||
// These are transient events that happen on every reconnect and carry no information
|
||||
// about persistent server-side failures. The reconnection logic in makeSocket handles
|
||||
// them independently.
|
||||
const WS_LIFECYCLE_STATUS_CODES = new Set([
|
||||
428, // connectionClosed
|
||||
408, // connectionLost / timedOut
|
||||
440, // connectionReplaced
|
||||
])
|
||||
|
||||
return new CircuitBreaker({
|
||||
name: 'connection-operations',
|
||||
failureThreshold: 3,
|
||||
failureWindow: 30000,
|
||||
resetTimeout: 60000,
|
||||
successThreshold: 1,
|
||||
shouldCountError: (error: Error) => {
|
||||
// CircuitTimeoutError messages embed the circuit name (e.g. "socket-query"), which
|
||||
// accidentally matches the "socket" pattern below and causes a self-reinforcing loop
|
||||
// where the breaker's own timeouts keep tripping the breaker. Exclude them explicitly.
|
||||
if (error instanceof CircuitTimeoutError) return false
|
||||
|
||||
// Exclude normal WS reconnect events (Connection Closed, Connection Lost, Timed Out, etc.).
|
||||
// These are not server-side failures — they happen on every restart/redeploy and should
|
||||
// never cause the circuit to open. If we counted them, the 3 concurrent parallel
|
||||
// post-login queries (sendPassiveIq, uploadPreKeysToServerIfRequired, digestKeyBundle) could all
|
||||
// fail simultaneously when the WS drops, instantly opening the circuit and blocking
|
||||
// profile-picture fetches and message delivery for the next 30 s.
|
||||
const statusCode = (error as { output?: { statusCode?: number } })?.output?.statusCode
|
||||
if (statusCode !== undefined && WS_LIFECYCLE_STATUS_CODES.has(statusCode)) return false
|
||||
|
||||
const message = error.message.toLowerCase()
|
||||
const code = (error as NodeJS.ErrnoException).code?.toLowerCase() || ''
|
||||
return connectionErrorPatterns.some(pattern => message.includes(pattern) || code.includes(pattern))
|
||||
},
|
||||
...customOptions
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a circuit breaker for message sending operations
|
||||
*/
|
||||
export function createMessageCircuitBreaker(customOptions?: Partial<CircuitBreakerOptions>): CircuitBreaker {
|
||||
return new CircuitBreaker({
|
||||
name: 'message-operations',
|
||||
failureThreshold: 5,
|
||||
failureWindow: 60000,
|
||||
resetTimeout: 15000,
|
||||
successThreshold: 2,
|
||||
timeout: 30000,
|
||||
...customOptions
|
||||
})
|
||||
}
|
||||
|
||||
export default CircuitBreaker
|
||||
@@ -6,12 +6,11 @@
|
||||
* @module Utils/health-status
|
||||
*/
|
||||
|
||||
import { globalCircuitRegistry } from './circuit-breaker.js'
|
||||
import { getVersionCacheStatus } from './version-cache.js'
|
||||
|
||||
/**
|
||||
* Circuit breaker health information (kept for k8s probe schema stability —
|
||||
* always reports an empty list now that circuit breakers were removed in
|
||||
* favour of bounded-retry).
|
||||
* Circuit breaker health information
|
||||
*/
|
||||
export interface CircuitBreakerHealth {
|
||||
name: string
|
||||
@@ -106,10 +105,44 @@ export function getHealthStatus(): HealthStatus {
|
||||
})
|
||||
}
|
||||
|
||||
// 2. Circuit breakers were removed in favour of bounded-retry.
|
||||
// Keep the field for backward-compat with k8s probe schema, always empty.
|
||||
// 2. Check circuit breakers
|
||||
const circuitBreakers: CircuitBreakerHealth[] = []
|
||||
checks.push({ name: 'circuit_breakers', status: 'pass' })
|
||||
let openCircuits = 0
|
||||
|
||||
for (const [name, breaker] of globalCircuitRegistry.getAll()) {
|
||||
const stats = breaker.getStats()
|
||||
const state = breaker.getState()
|
||||
|
||||
circuitBreakers.push({
|
||||
name,
|
||||
state,
|
||||
failures: stats.totalFailures,
|
||||
successes: stats.totalSuccesses,
|
||||
totalCalls: stats.totalCalls
|
||||
})
|
||||
|
||||
if (state === 'open') {
|
||||
openCircuits++
|
||||
}
|
||||
}
|
||||
|
||||
if (openCircuits > 0) {
|
||||
checks.push({
|
||||
name: 'circuit_breakers',
|
||||
status: openCircuits > 2 ? 'fail' : 'warn',
|
||||
message: `${openCircuits} circuit breaker(s) are open`
|
||||
})
|
||||
if (openCircuits > 2) {
|
||||
overallStatus = 'unhealthy'
|
||||
} else if (overallStatus === 'healthy') {
|
||||
overallStatus = 'degraded'
|
||||
}
|
||||
} else {
|
||||
checks.push({
|
||||
name: 'circuit_breakers',
|
||||
status: 'pass'
|
||||
})
|
||||
}
|
||||
|
||||
// 3. Check memory usage (warn if > 90%)
|
||||
const memUsage = process.memoryUsage()
|
||||
|
||||
+1
-1
@@ -33,8 +33,8 @@ export * from './trace-context'
|
||||
export * from './prometheus-metrics'
|
||||
|
||||
// Resilience and performance
|
||||
export * from './bounded-retry'
|
||||
export * from './cache-utils'
|
||||
export * from './circuit-breaker'
|
||||
export * from './retry-utils'
|
||||
|
||||
// Telemetry and detection mitigation
|
||||
|
||||
@@ -1570,6 +1570,26 @@ export const metrics = {
|
||||
new Counter('socket_reconnects_total', 'Total socket reconnection attempts', ['reason'])
|
||||
),
|
||||
|
||||
// ========== Circuit Breaker Metrics ==========
|
||||
circuitBreakerState: baileysMetrics.register(
|
||||
new Gauge('circuit_breaker_state', 'Circuit breaker state (0=closed, 1=open, 2=half-open)', ['name'])
|
||||
),
|
||||
circuitBreakerTrips: baileysMetrics.register(
|
||||
new Counter('circuit_breaker_trips_total', 'Total circuit breaker trips', ['name'])
|
||||
),
|
||||
circuitBreakerRecoveries: baileysMetrics.register(
|
||||
new Counter('circuit_breaker_recoveries_total', 'Total circuit breaker recoveries', ['name'])
|
||||
),
|
||||
circuitBreakerRejections: baileysMetrics.register(
|
||||
new Counter('circuit_breaker_rejections_total', 'Total requests rejected by circuit breaker', ['name'])
|
||||
),
|
||||
circuitBreakerSuccesses: baileysMetrics.register(
|
||||
new Counter('circuit_breaker_successes_total', 'Total successful requests through circuit breaker', ['name'])
|
||||
),
|
||||
circuitBreakerFailures: baileysMetrics.register(
|
||||
new Counter('circuit_breaker_failures_total', 'Total failed requests through circuit breaker', ['name'])
|
||||
),
|
||||
|
||||
// ========== Encryption Metrics ==========
|
||||
encryptionOperations: baileysMetrics.register(
|
||||
new Counter('encryption_operations_total', 'Total encryption operations', ['operation'])
|
||||
@@ -2147,6 +2167,9 @@ function initializeMetricsWithLabels(): void {
|
||||
metrics.messageFailures?.inc({ type: 'text', reason: 'max_retries' }, 0)
|
||||
metrics.messageFailures?.inc({ type: 'other', reason: 'max_retries' }, 0)
|
||||
|
||||
// Circuit breaker metric
|
||||
metrics.circuitBreakerTrips?.inc({ name: 'main' }, 0)
|
||||
|
||||
console.log('[Prometheus] Initialized metrics with labels')
|
||||
} catch (error) {
|
||||
console.error('[Prometheus] Error initializing metrics with labels:', error)
|
||||
|
||||
+148
-13
@@ -6,21 +6,15 @@
|
||||
* - Jitter to avoid thundering herd
|
||||
* - Configurable max attempts
|
||||
* - Customizable retry predicates
|
||||
* - Cancellation support (AbortSignal)
|
||||
*
|
||||
* Used by `decode-wa-message.ts` for Bad MAC retry recovery on the decrypt
|
||||
* path. For socket-operation retries (uploadPreKeys etc.) prefer
|
||||
* `withBoundedRetry` from `./bounded-retry.ts`, which is empirically
|
||||
* aligned with WhatsApp Android's per-operation backoff.
|
||||
*
|
||||
* NOTE: the previous `circuitBreaker` integration option, the `withRetry`
|
||||
* decorator, the `retryable` wrapper and the `RetryManager` class were
|
||||
* removed when the socket-level circuit breaker was retired — see PR #393
|
||||
* for the rationale.
|
||||
* - Circuit breaker integration
|
||||
* - Event hooks
|
||||
* - Cancellation support
|
||||
*
|
||||
* @module Utils/retry-utils
|
||||
*/
|
||||
|
||||
import { EventEmitter } from 'events'
|
||||
import type { CircuitBreaker } from './circuit-breaker.js'
|
||||
import { metrics } from './prometheus-metrics.js'
|
||||
|
||||
/**
|
||||
@@ -65,6 +59,8 @@ export interface RetryOptions {
|
||||
operationName?: string
|
||||
/** Collect metrics */
|
||||
collectMetrics?: boolean
|
||||
/** Circuit breaker for integration */
|
||||
circuitBreaker?: CircuitBreaker
|
||||
/** Callback before each retry */
|
||||
onRetry?: (error: Error, attempt: number, delay: number) => void | Promise<void>
|
||||
/** Callback on success */
|
||||
@@ -287,6 +283,7 @@ export async function retry<T>(
|
||||
timeout: options.timeout,
|
||||
operationName: options.operationName ?? 'operation',
|
||||
collectMetrics: options.collectMetrics ?? true,
|
||||
circuitBreaker: options.circuitBreaker,
|
||||
onRetry: options.onRetry ?? (() => {}),
|
||||
onSuccess: options.onSuccess ?? (() => {}),
|
||||
onFailure: options.onFailure ?? (() => {}),
|
||||
@@ -316,6 +313,11 @@ export async function retry<T>(
|
||||
throw new RetryAbortedError(attempt)
|
||||
}
|
||||
|
||||
// Check circuit breaker
|
||||
if (config.circuitBreaker?.isOpen()) {
|
||||
throw new Error(`Circuit breaker "${config.circuitBreaker.getName()}" is open`)
|
||||
}
|
||||
|
||||
try {
|
||||
// Execute operation
|
||||
let result: T
|
||||
@@ -423,6 +425,138 @@ export function createRetrier(defaultOptions: RetryOptions = {}) {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Decorator to add retry to method
|
||||
*/
|
||||
export function withRetry(options: RetryOptions = {}) {
|
||||
return function (
|
||||
_target: unknown,
|
||||
propertyKey: string,
|
||||
descriptor: TypedPropertyDescriptor<(...args: unknown[]) => unknown>
|
||||
) {
|
||||
const originalMethod = descriptor.value
|
||||
if (!originalMethod) return descriptor
|
||||
|
||||
descriptor.value = async function (...args: unknown[]): Promise<unknown> {
|
||||
return retry(() => originalMethod.apply(this, args), {
|
||||
...options,
|
||||
operationName: options.operationName || propertyKey
|
||||
})
|
||||
}
|
||||
|
||||
return descriptor
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Wrapper for function with retry
|
||||
*/
|
||||
// eslint-disable-next-line space-before-function-paren
|
||||
export function retryable<T extends (...args: unknown[]) => unknown>(
|
||||
fn: T,
|
||||
options: RetryOptions = {}
|
||||
): (...args: Parameters<T>) => Promise<ReturnType<T>> {
|
||||
return async (...args: Parameters<T>): Promise<ReturnType<T>> => {
|
||||
return retry(() => fn(...args), options) as Promise<ReturnType<T>>
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Class to manage retries with state
|
||||
*/
|
||||
export class RetryManager extends EventEmitter {
|
||||
private activeRetries: Map<string, { cancel: () => void; context: RetryContext }> = new Map()
|
||||
private defaultOptions: RetryOptions
|
||||
|
||||
constructor(defaultOptions: RetryOptions = {}) {
|
||||
super()
|
||||
this.defaultOptions = defaultOptions
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute operation with retry
|
||||
*/
|
||||
async execute<T>(
|
||||
id: string,
|
||||
operation: (context: RetryContext) => T | Promise<T>,
|
||||
options?: RetryOptions
|
||||
): Promise<T> {
|
||||
// Cancel previous retry with same ID
|
||||
this.cancel(id)
|
||||
|
||||
const abortController = new AbortController()
|
||||
const mergedOptions = { ...this.defaultOptions, ...options, abortSignal: abortController.signal }
|
||||
|
||||
const retryPromise = retry(context => {
|
||||
this.activeRetries.set(id, {
|
||||
cancel: () => abortController.abort(),
|
||||
context
|
||||
})
|
||||
this.emit('attempt', { id, attempt: context.attempt })
|
||||
return operation(context)
|
||||
}, mergedOptions)
|
||||
|
||||
try {
|
||||
const result = await retryPromise
|
||||
this.emit('success', { id })
|
||||
return result
|
||||
} catch (error) {
|
||||
this.emit('failure', { id, error })
|
||||
throw error
|
||||
} finally {
|
||||
this.activeRetries.delete(id)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel in-progress retry
|
||||
*/
|
||||
cancel(id: string): boolean {
|
||||
const active = this.activeRetries.get(id)
|
||||
if (active) {
|
||||
active.cancel()
|
||||
this.activeRetries.delete(id)
|
||||
this.emit('cancelled', { id })
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel all retries
|
||||
*/
|
||||
cancelAll(): void {
|
||||
for (const [id, active] of this.activeRetries) {
|
||||
active.cancel()
|
||||
this.emit('cancelled', { id })
|
||||
}
|
||||
|
||||
this.activeRetries.clear()
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if there is an active retry
|
||||
*/
|
||||
isActive(id: string): boolean {
|
||||
return this.activeRetries.has(id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Return active retry context
|
||||
*/
|
||||
getContext(id: string): RetryContext | undefined {
|
||||
return this.activeRetries.get(id)?.context
|
||||
}
|
||||
|
||||
/**
|
||||
* Return active retry IDs
|
||||
*/
|
||||
getActiveIds(): string[] {
|
||||
return Array.from(this.activeRetries.keys())
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Common predicates for shouldRetry
|
||||
*/
|
||||
@@ -526,8 +660,9 @@ export const retryConfigs = {
|
||||
* Uses fixed delay array: 1s, 2s, 5s, 10s, 20s (with ±15% jitter)
|
||||
*
|
||||
* NOTE: Values are hardcoded instead of referencing RETRY_BACKOFF_DELAYS/RETRY_JITTER_FACTOR
|
||||
* to prevent "Cannot access before initialization" errors in ESM environments
|
||||
* caused by indirect circular imports through prometheus-metrics.ts.
|
||||
* to prevent "Cannot access before initialization" errors in ESM environments.
|
||||
* This occurs when modules are loaded in specific orders due to indirect circular imports
|
||||
* (e.g., via prometheus-metrics.ts -> circuit-breaker.ts chain).
|
||||
* Keep these values in sync with the constants above (lines 25, 31).
|
||||
*/
|
||||
rsocket: {
|
||||
|
||||
@@ -6,9 +6,7 @@
|
||||
* - Configurable log levels (trace, debug, info, warn, error, fatal)
|
||||
* - JSON formatting for log analysis
|
||||
* - Hierarchical context with child loggers
|
||||
* - External system integration via fire-and-forget hooks (failures
|
||||
* counted in hookFailures metric; no per-hook circuit breaker — chronic
|
||||
* breakage is visible to operators via the metric)
|
||||
* - External system integration via hooks with circuit breaker
|
||||
* - Logging metrics with Prometheus integration
|
||||
* - Sensitive data sanitization
|
||||
* - Log buffering for batch writes
|
||||
@@ -75,6 +73,10 @@ export interface StructuredLoggerConfig {
|
||||
maxLogsPerSecond?: number
|
||||
/** Enable async logging queue (default: false) */
|
||||
enableAsyncQueue?: boolean
|
||||
/** Circuit breaker failure threshold for external hooks (default: 5) */
|
||||
circuitBreakerThreshold?: number
|
||||
/** Circuit breaker reset timeout in ms (default: 30000) */
|
||||
circuitBreakerResetMs?: number
|
||||
/** Enable Prometheus metrics integration (default: false) */
|
||||
enableMetrics?: boolean
|
||||
}
|
||||
@@ -153,6 +155,8 @@ export interface LoggerMetrics {
|
||||
bufferFlushes: number
|
||||
/** External hook failures */
|
||||
hookFailures: number
|
||||
/** Circuit breaker trips */
|
||||
circuitBreakerTrips: number
|
||||
/** Average log processing time in ms */
|
||||
avgProcessingTimeMs: number
|
||||
}
|
||||
@@ -165,6 +169,8 @@ export interface LoggerStatistics extends LoggerMetrics {
|
||||
bufferSize: number
|
||||
/** Rate limiter tokens available */
|
||||
rateLimiterTokens: number
|
||||
/** Circuit breaker state */
|
||||
circuitBreakerState: 'closed' | 'open' | 'half-open'
|
||||
/** Queue size (if async enabled) */
|
||||
queueSize: number
|
||||
/** Created timestamp */
|
||||
@@ -236,6 +242,69 @@ class RateLimiter {
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// CIRCUIT BREAKER
|
||||
// ============================================================================
|
||||
|
||||
/**
|
||||
* Circuit breaker for external hook protection
|
||||
*/
|
||||
class CircuitBreaker {
|
||||
private failures = 0
|
||||
private lastFailure = 0
|
||||
private state: 'closed' | 'open' | 'half-open' = 'closed'
|
||||
|
||||
constructor(
|
||||
private readonly threshold: number,
|
||||
private readonly resetTimeoutMs: number
|
||||
) {}
|
||||
|
||||
async execute<T>(operation: () => Promise<T>): Promise<T | null> {
|
||||
if (this.state === 'open') {
|
||||
if (Date.now() - this.lastFailure > this.resetTimeoutMs) {
|
||||
this.state = 'half-open'
|
||||
} else {
|
||||
return null // Circuit is open, skip
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await operation()
|
||||
this.onSuccess()
|
||||
return result
|
||||
} catch (error) {
|
||||
this.onFailure()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private onSuccess(): void {
|
||||
this.failures = 0
|
||||
this.state = 'closed'
|
||||
}
|
||||
|
||||
private onFailure(): void {
|
||||
this.failures++
|
||||
this.lastFailure = Date.now()
|
||||
if (this.failures >= this.threshold) {
|
||||
this.state = 'open'
|
||||
}
|
||||
}
|
||||
|
||||
getState(): 'closed' | 'open' | 'half-open' {
|
||||
// Check if should transition from open to half-open
|
||||
if (this.state === 'open' && Date.now() - this.lastFailure > this.resetTimeoutMs) {
|
||||
this.state = 'half-open'
|
||||
}
|
||||
|
||||
return this.state
|
||||
}
|
||||
|
||||
getFailures(): number {
|
||||
return this.failures
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// ASYNC LOG QUEUE
|
||||
// ============================================================================
|
||||
@@ -334,6 +403,8 @@ export class StructuredLogger implements ILogger {
|
||||
private rateLimiter: RateLimiter | null = null
|
||||
|
||||
// Circuit breaker for external hook
|
||||
private circuitBreaker: CircuitBreaker | null = null
|
||||
|
||||
// Async queue
|
||||
private asyncQueue: AsyncLogQueue | null = null
|
||||
|
||||
@@ -364,6 +435,8 @@ export class StructuredLogger implements ILogger {
|
||||
enableRateLimiting: config.enableRateLimiting ?? envConfig.enableRateLimiting ?? false,
|
||||
maxLogsPerSecond: config.maxLogsPerSecond ?? envConfig.maxLogsPerSecond ?? 1000,
|
||||
enableAsyncQueue: config.enableAsyncQueue ?? envConfig.enableAsyncQueue ?? false,
|
||||
circuitBreakerThreshold: config.circuitBreakerThreshold ?? 5,
|
||||
circuitBreakerResetMs: config.circuitBreakerResetMs ?? 30000,
|
||||
enableMetrics: config.enableMetrics ?? envConfig.enableMetrics ?? false
|
||||
}
|
||||
|
||||
@@ -382,6 +455,7 @@ export class StructuredLogger implements ILogger {
|
||||
droppedLogs: 0,
|
||||
bufferFlushes: 0,
|
||||
hookFailures: 0,
|
||||
circuitBreakerTrips: 0,
|
||||
avgProcessingTimeMs: 0
|
||||
}
|
||||
|
||||
@@ -390,6 +464,11 @@ export class StructuredLogger implements ILogger {
|
||||
this.rateLimiter = new RateLimiter(this.config.maxLogsPerSecond)
|
||||
}
|
||||
|
||||
// Initialize circuit breaker for external hook
|
||||
if (this.config.externalHook) {
|
||||
this.circuitBreaker = new CircuitBreaker(this.config.circuitBreakerThreshold, this.config.circuitBreakerResetMs)
|
||||
}
|
||||
|
||||
// Initialize async queue
|
||||
if (this.config.enableAsyncQueue) {
|
||||
this.asyncQueue = new AsyncLogQueue()
|
||||
@@ -490,27 +569,19 @@ export class StructuredLogger implements ILogger {
|
||||
this.output(entry)
|
||||
}
|
||||
|
||||
// External hook — fire-and-forget. Failures are counted in
|
||||
// hookFailures metric so chronic breakage is visible to operators
|
||||
// who can fix their hook. We do not gate further calls — that is
|
||||
// the consumer's responsibility (consistent with the rest of the
|
||||
// codebase after the circuit-breaker removal).
|
||||
// External hook with circuit breaker
|
||||
const externalHook = this.config.externalHook
|
||||
if (externalHook) {
|
||||
try {
|
||||
const hookResult = externalHook(entry) as unknown
|
||||
// Use Promise.resolve().catch() so any thenable (including
|
||||
// Promise-likes that only implement .then) is handled
|
||||
// safely — calling .catch directly would throw on such
|
||||
// objects (Copilot review on PR #393).
|
||||
if (hookResult && typeof (hookResult as PromiseLike<unknown>).then === 'function') {
|
||||
void Promise.resolve(hookResult as PromiseLike<unknown>).catch(() => {
|
||||
this.metrics.hookFailures++
|
||||
})
|
||||
}
|
||||
} catch {
|
||||
this.metrics.hookFailures++
|
||||
}
|
||||
if (externalHook && this.circuitBreaker) {
|
||||
this.circuitBreaker
|
||||
.execute(async () => {
|
||||
await Promise.resolve(externalHook(entry))
|
||||
})
|
||||
.catch(() => {
|
||||
this.metrics.hookFailures++
|
||||
if (this.circuitBreaker?.getState() === 'open') {
|
||||
this.metrics.circuitBreakerTrips++
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Track processing time
|
||||
@@ -763,6 +834,7 @@ export class StructuredLogger implements ILogger {
|
||||
...this.metrics,
|
||||
bufferSize: this.buffer.length,
|
||||
rateLimiterTokens: this.rateLimiter?.getTokens() ?? 0,
|
||||
circuitBreakerState: this.circuitBreaker?.getState() ?? 'closed',
|
||||
queueSize: this.asyncQueue?.getSize() ?? 0,
|
||||
createdAt: this.createdAt,
|
||||
uptimeMs: Date.now() - this.createdAt
|
||||
@@ -788,6 +860,7 @@ export class StructuredLogger implements ILogger {
|
||||
droppedLogs: 0,
|
||||
bufferFlushes: 0,
|
||||
hookFailures: 0,
|
||||
circuitBreakerTrips: 0,
|
||||
avgProcessingTimeMs: 0
|
||||
}
|
||||
this.totalProcessingTime = 0
|
||||
@@ -820,6 +893,7 @@ export class StructuredLogger implements ILogger {
|
||||
|
||||
// Clear references
|
||||
this.rateLimiter = null
|
||||
this.circuitBreaker = null
|
||||
this.metricsModule = null
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
*/
|
||||
|
||||
import type { BinaryNode } from '../WABinary/types.js'
|
||||
import { CircuitBreaker, CircuitOpenError, createConnectionCircuitBreaker } from './circuit-breaker.js'
|
||||
import type { ILogger } from './logger.js'
|
||||
import { metrics } from './prometheus-metrics.js'
|
||||
|
||||
@@ -57,6 +58,8 @@ export interface UnifiedSessionOptions {
|
||||
enabled?: boolean
|
||||
/** Logger instance for debugging */
|
||||
logger?: ILogger
|
||||
/** Enable circuit breaker protection for send operations */
|
||||
enableCircuitBreaker?: boolean
|
||||
/** Function to send binary nodes to WhatsApp */
|
||||
sendNode?: (node: BinaryNode) => Promise<void>
|
||||
}
|
||||
@@ -90,9 +93,7 @@ export type UnifiedSessionTrigger = 'login' | 'pairing' | 'presence' | 'manual'
|
||||
* Manages the unified_session telemetry feature with:
|
||||
* - Server time synchronization
|
||||
* - Rate limiting (prevents spam)
|
||||
* - Best-effort send semantics (failures are swallowed and counted —
|
||||
* telemetry is non-critical so a single attempt with the underlying
|
||||
* sendNode timeout is enough; no retry, no circuit breaker)
|
||||
* - Circuit breaker protection
|
||||
* - Prometheus metrics integration
|
||||
* - Structured logging
|
||||
*
|
||||
@@ -120,6 +121,7 @@ export class UnifiedSessionManager {
|
||||
}
|
||||
|
||||
private readonly options: Required<UnifiedSessionOptions>
|
||||
private circuitBreaker: CircuitBreaker | null = null
|
||||
|
||||
/** Minimum interval between unified_session sends (1 minute) */
|
||||
private static readonly MIN_SEND_INTERVAL_MS = TimeMs.Minute
|
||||
@@ -128,6 +130,7 @@ export class UnifiedSessionManager {
|
||||
this.options = {
|
||||
enabled: options.enabled ?? true,
|
||||
logger: options.logger ?? (console as unknown as ILogger),
|
||||
enableCircuitBreaker: options.enableCircuitBreaker ?? true,
|
||||
sendNode:
|
||||
options.sendNode ??
|
||||
(async () => {
|
||||
@@ -135,6 +138,25 @@ export class UnifiedSessionManager {
|
||||
})
|
||||
}
|
||||
|
||||
// Initialize circuit breaker if enabled
|
||||
if (this.options.enableCircuitBreaker) {
|
||||
this.circuitBreaker = createConnectionCircuitBreaker({
|
||||
name: 'unified-session',
|
||||
failureThreshold: 3,
|
||||
failureWindow: 60000,
|
||||
resetTimeout: 30000,
|
||||
successThreshold: 1,
|
||||
timeout: 10000,
|
||||
onStateChange: (from, to) => {
|
||||
this.options.logger.debug?.({ from, to }, 'Unified session circuit breaker state changed')
|
||||
},
|
||||
onOpen: () => {
|
||||
this.options.logger.warn?.('Unified session circuit breaker OPENED')
|
||||
metrics.circuitBreakerTrips?.inc({ name: 'unified-session' })
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
this.state.isInitialized = true
|
||||
this.options.logger.debug?.('UnifiedSessionManager initialized')
|
||||
}
|
||||
@@ -270,12 +292,20 @@ export class UnifiedSessionManager {
|
||||
}
|
||||
|
||||
try {
|
||||
// Telemetry is non-critical: a single send attempt with the
|
||||
// underlying sendNode timeout is enough. No retry/circuit needed.
|
||||
await sendOperation()
|
||||
// Execute with circuit breaker if available
|
||||
if (this.circuitBreaker) {
|
||||
await this.circuitBreaker.execute(sendOperation)
|
||||
} else {
|
||||
await sendOperation()
|
||||
}
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
this.options.logger.warn?.({ trigger, error: errorMessage }, 'Failed to send unified session telemetry')
|
||||
|
||||
if (error instanceof CircuitOpenError) {
|
||||
this.options.logger.warn?.({ trigger, circuitState: error.state }, 'Unified session blocked by circuit breaker')
|
||||
} else {
|
||||
this.options.logger.warn?.({ trigger, error: errorMessage }, 'Failed to send unified session telemetry')
|
||||
}
|
||||
|
||||
// Record failure metric
|
||||
metrics.errors?.inc({ category: 'unified_session', code: 'send_failed' })
|
||||
@@ -301,6 +331,7 @@ export class UnifiedSessionManager {
|
||||
sendCount: 0,
|
||||
isInitialized: true
|
||||
}
|
||||
this.circuitBreaker?.reset()
|
||||
this.options.logger.debug?.('UnifiedSessionManager reset')
|
||||
}
|
||||
|
||||
@@ -308,6 +339,7 @@ export class UnifiedSessionManager {
|
||||
* Destroy the manager and clean up resources.
|
||||
*/
|
||||
destroy(): void {
|
||||
this.circuitBreaker?.destroy()
|
||||
this.state.isInitialized = false
|
||||
this.options.logger.debug?.('UnifiedSessionManager destroyed')
|
||||
}
|
||||
|
||||
@@ -1,391 +0,0 @@
|
||||
import { jest } from '@jest/globals'
|
||||
import {
|
||||
withBoundedRetry,
|
||||
BoundedRetryGiveUpError,
|
||||
BoundedRetryAbortedError,
|
||||
WHATSAPP_BACKOFF_DELAYS,
|
||||
DEFAULT_TTL_MS
|
||||
} from '../../Utils/bounded-retry'
|
||||
|
||||
// All tests use very short real delays (1-50ms) so they run fast.
|
||||
// This sidesteps the fragility of jest fake timers + async chains.
|
||||
|
||||
describe('bounded-retry — WhatsApp-aligned per-operation retry', () => {
|
||||
test('default delay sequence matches WhatsApp Android empirical capture', () => {
|
||||
// Frida trace: 3000 -> 10000 -> 60000 -> ~64000 -> 120000 ms
|
||||
expect(WHATSAPP_BACKOFF_DELAYS).toEqual([3000, 10000, 60000, 60000, 120000])
|
||||
})
|
||||
|
||||
test('default TTL is 10 minutes (matches empirical stabilisation + safety buffer)', () => {
|
||||
expect(DEFAULT_TTL_MS).toBe(600_000)
|
||||
})
|
||||
|
||||
test('first-attempt success returns immediately, no retries', async () => {
|
||||
const op = jest.fn<() => Promise<string>>().mockResolvedValue('ok')
|
||||
const result = await withBoundedRetry(op, { name: 'fast-success' })
|
||||
expect(result).toBe('ok')
|
||||
expect(op).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
test('retries with the configured delay sequence after failures', async () => {
|
||||
const op = jest
|
||||
.fn<() => Promise<number>>()
|
||||
.mockRejectedValueOnce(new Error('try1'))
|
||||
.mockRejectedValueOnce(new Error('try2'))
|
||||
.mockResolvedValueOnce(42)
|
||||
|
||||
const onRetry = jest.fn<(err: Error, attempt: number, delayMs: number) => void>()
|
||||
const result = await withBoundedRetry(op, {
|
||||
name: 'three-tries',
|
||||
delays: [10, 20, 40],
|
||||
jitter: 0,
|
||||
onRetry
|
||||
})
|
||||
|
||||
expect(result).toBe(42)
|
||||
expect(op).toHaveBeenCalledTimes(3)
|
||||
expect(onRetry).toHaveBeenCalledTimes(2)
|
||||
expect(onRetry.mock.calls[0]![2]).toBe(10)
|
||||
expect(onRetry.mock.calls[1]![2]).toBe(20)
|
||||
})
|
||||
|
||||
test('TTL exhaustion throws BoundedRetryGiveUpError (no infinite retries)', async () => {
|
||||
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('always-fails'))
|
||||
|
||||
await expect(
|
||||
withBoundedRetry(op, {
|
||||
name: 'ttl-test',
|
||||
delays: [10],
|
||||
jitter: 0,
|
||||
ttlMs: 50
|
||||
})
|
||||
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
|
||||
|
||||
// Should have tried multiple times before giving up
|
||||
expect(op.mock.calls.length).toBeGreaterThanOrEqual(2)
|
||||
})
|
||||
|
||||
test('per-attempt timeout fires for slow operations', async () => {
|
||||
const slowOp = jest.fn<() => Promise<string>>(() => new Promise(() => {})) // hangs
|
||||
|
||||
await expect(
|
||||
withBoundedRetry(slowOp, {
|
||||
name: 'slow',
|
||||
delays: [5],
|
||||
jitter: 0,
|
||||
ttlMs: 200,
|
||||
perAttemptTimeoutMs: 30
|
||||
})
|
||||
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
|
||||
|
||||
// Multiple attempts should have been tried (each timed out at 30ms)
|
||||
expect(slowOp.mock.calls.length).toBeGreaterThan(1)
|
||||
})
|
||||
|
||||
test('shouldRetry predicate can short-circuit retries', async () => {
|
||||
const op = jest
|
||||
.fn<() => Promise<string>>()
|
||||
.mockRejectedValueOnce(new Error('transient'))
|
||||
.mockRejectedValueOnce(new Error('FATAL: do not retry'))
|
||||
|
||||
await expect(
|
||||
withBoundedRetry(op, {
|
||||
name: 'predicate-test',
|
||||
delays: [1],
|
||||
jitter: 0,
|
||||
shouldRetry: err => !err.message.startsWith('FATAL')
|
||||
})
|
||||
).rejects.toThrow('FATAL: do not retry')
|
||||
|
||||
expect(op).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
test('AbortSignal cancels in-flight retry sleep', async () => {
|
||||
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('fail'))
|
||||
const ctrl = new AbortController()
|
||||
|
||||
const promise = withBoundedRetry(op, {
|
||||
name: 'abort-test',
|
||||
delays: [10_000], // long sleep so abort can happen during it
|
||||
jitter: 0,
|
||||
signal: ctrl.signal
|
||||
})
|
||||
|
||||
// First attempt fires immediately; sleep starts
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
ctrl.abort()
|
||||
|
||||
await expect(promise).rejects.toBeInstanceOf(BoundedRetryAbortedError)
|
||||
})
|
||||
|
||||
test('two concurrent retry chains do NOT interfere with each other (per-op isolation)', async () => {
|
||||
// Key advantage over circuit breaker: failures in one chain do not
|
||||
// block another chain. With circuit breaker, 5 failures from chain A
|
||||
// would block chain B too. With bounded-retry, B is independent.
|
||||
|
||||
const flakyOp = jest
|
||||
.fn<() => Promise<string>>()
|
||||
.mockRejectedValueOnce(new Error('a1'))
|
||||
.mockResolvedValueOnce('a-ok')
|
||||
|
||||
const fastOp = jest.fn<() => Promise<string>>().mockResolvedValue('b-ok')
|
||||
|
||||
// Start a (will retry once after 30ms) and b (immediate)
|
||||
const aPromise = withBoundedRetry(flakyOp, { name: 'a', delays: [30], jitter: 0 })
|
||||
const bPromise = withBoundedRetry(fastOp, { name: 'b', delays: [30], jitter: 0 })
|
||||
|
||||
// b should resolve right away
|
||||
const bResult = await bPromise
|
||||
expect(bResult).toBe('b-ok')
|
||||
expect(fastOp).toHaveBeenCalledTimes(1)
|
||||
|
||||
// a still running — wait for it
|
||||
const aResult = await aPromise
|
||||
expect(aResult).toBe('a-ok')
|
||||
expect(flakyOp).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
test('delay sequence caps at last value (no unbounded growth)', async () => {
|
||||
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('fail'))
|
||||
const onRetry = jest.fn<(err: Error, attempt: number, delayMs: number) => void>()
|
||||
|
||||
await expect(
|
||||
withBoundedRetry(op, {
|
||||
name: 'cap-test',
|
||||
delays: [5, 10, 15], // 3 distinct steps; 4th+ retry should reuse 15
|
||||
jitter: 0,
|
||||
ttlMs: 300,
|
||||
onRetry
|
||||
})
|
||||
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
|
||||
|
||||
// Find delays >= 4th retry — all should be capped at 15
|
||||
const lateRetries = onRetry.mock.calls.filter(c => c[1] >= 4)
|
||||
expect(lateRetries.length).toBeGreaterThan(0)
|
||||
for (const call of lateRetries) {
|
||||
expect(call[2]).toBeLessThanOrEqual(15)
|
||||
}
|
||||
})
|
||||
|
||||
test('jitter randomises delay within +/- factor', async () => {
|
||||
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('fail'))
|
||||
const observed: number[] = []
|
||||
|
||||
// Single run, capture multiple onRetry calls (each with different jitter)
|
||||
await withBoundedRetry(op, {
|
||||
name: 'jitter-test',
|
||||
delays: [50], // base delay
|
||||
jitter: 0.3, // +/- 30% -> [35, 65]
|
||||
ttlMs: 1000, // ttl high enough for ~10 retries
|
||||
perAttemptTimeoutMs: 5,
|
||||
onRetry: (_, __, d) => observed.push(d)
|
||||
}).catch(() => {})
|
||||
|
||||
expect(observed.length).toBeGreaterThanOrEqual(3)
|
||||
const unique = new Set(observed)
|
||||
expect(unique.size).toBeGreaterThan(1)
|
||||
// Every observed delay must be within jitter window — but the LAST one
|
||||
// may be capped by remaining ttl budget, so check all but the last
|
||||
const inWindow = observed.slice(0, -1)
|
||||
for (const d of inWindow) {
|
||||
expect(d).toBeGreaterThanOrEqual(35)
|
||||
expect(d).toBeLessThanOrEqual(65)
|
||||
}
|
||||
})
|
||||
|
||||
test('delay does not exceed remaining TTL budget', async () => {
|
||||
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('fail'))
|
||||
const onRetry = jest.fn<(err: Error, attempt: number, delayMs: number) => void>()
|
||||
|
||||
await withBoundedRetry(op, {
|
||||
name: 'budget-cap',
|
||||
delays: [200], // would exceed ttl
|
||||
jitter: 0,
|
||||
ttlMs: 50, // very short
|
||||
onRetry
|
||||
}).catch(() => {})
|
||||
|
||||
// All scheduled delays should be capped to remaining budget
|
||||
for (const call of onRetry.mock.calls) {
|
||||
expect(call[2]).toBeLessThanOrEqual(50)
|
||||
}
|
||||
})
|
||||
|
||||
// ─── Tests for fixes from PR #393 review ──────────────────────────────
|
||||
|
||||
test('TTL is enforced BEFORE the next attempt (not just after)', async () => {
|
||||
// Regression: previously TTL was only checked after a failure, so a
|
||||
// new attempt could start with 0ms remaining and run for the full
|
||||
// per-attempt timeout, overshooting wall-clock budget by that much.
|
||||
const slowOp = jest.fn<() => Promise<string>>(() => new Promise(resolve => setTimeout(() => resolve('late'), 200)))
|
||||
|
||||
const start = Date.now()
|
||||
await expect(
|
||||
withBoundedRetry(slowOp, {
|
||||
name: 'ttl-strict',
|
||||
delays: [10],
|
||||
jitter: 0,
|
||||
ttlMs: 50,
|
||||
perAttemptTimeoutMs: 1000 // huge, must be capped by TTL
|
||||
})
|
||||
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
|
||||
|
||||
const elapsed = Date.now() - start
|
||||
// With strict TTL enforcement, total runtime should be very close to ttlMs.
|
||||
// Allow up to 100ms slack for Node timer + microtask scheduling.
|
||||
expect(elapsed).toBeLessThan(150)
|
||||
})
|
||||
|
||||
test('per-attempt timeout is capped by remaining TTL budget', async () => {
|
||||
// If user passes perAttemptTimeoutMs > ttlMs, the cap should apply.
|
||||
const slowOp = jest.fn<() => Promise<string>>(() => new Promise(() => {})) // hangs forever
|
||||
|
||||
const start = Date.now()
|
||||
await expect(
|
||||
withBoundedRetry(slowOp, {
|
||||
name: 'attempt-cap',
|
||||
delays: [1],
|
||||
jitter: 0,
|
||||
ttlMs: 30,
|
||||
perAttemptTimeoutMs: 5000 // way bigger than ttl
|
||||
})
|
||||
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
|
||||
|
||||
const elapsed = Date.now() - start
|
||||
expect(elapsed).toBeLessThan(150) // not 5000ms!
|
||||
})
|
||||
|
||||
test('operation receives an AbortSignal that is aborted on per-attempt timeout', async () => {
|
||||
// New API: operation can opt into cancellation via the signal arg.
|
||||
const aborts: boolean[] = []
|
||||
const opThatRespectsSignal = jest.fn(
|
||||
(signal?: AbortSignal) =>
|
||||
new Promise<string>((resolve, reject) => {
|
||||
const timer = setTimeout(() => resolve('late'), 100)
|
||||
signal?.addEventListener('abort', () => {
|
||||
clearTimeout(timer)
|
||||
aborts.push(true)
|
||||
reject(new Error('aborted by signal'))
|
||||
})
|
||||
})
|
||||
)
|
||||
|
||||
await withBoundedRetry(opThatRespectsSignal, {
|
||||
name: 'signal-aware',
|
||||
delays: [1],
|
||||
jitter: 0,
|
||||
ttlMs: 80,
|
||||
perAttemptTimeoutMs: 20 // smaller than 100ms op duration
|
||||
}).catch(() => {})
|
||||
|
||||
expect(aborts.length).toBeGreaterThan(0) // at least one attempt was aborted
|
||||
})
|
||||
|
||||
test('sleep listener is removed when timeout completes normally (no leak)', async () => {
|
||||
// Regression: sleep used signal.addEventListener with { once: true } but
|
||||
// never removed the listener on timer-resolve. With many retries on a
|
||||
// shared signal, listeners would accumulate.
|
||||
//
|
||||
// Make the failure mode observable by spying on add/removeEventListener
|
||||
// directly so the assertion fails if cleanup is removed.
|
||||
const ctrl = new AbortController()
|
||||
const realAdd = ctrl.signal.addEventListener.bind(ctrl.signal)
|
||||
const realRemove = ctrl.signal.removeEventListener.bind(ctrl.signal)
|
||||
let addCount = 0
|
||||
let removeCount = 0
|
||||
ctrl.signal.addEventListener = ((type: string, listener: never, opts?: never) => {
|
||||
if (type === 'abort') addCount++
|
||||
return realAdd(type, listener, opts)
|
||||
}) as typeof ctrl.signal.addEventListener
|
||||
ctrl.signal.removeEventListener = ((type: string, listener: never, opts?: never) => {
|
||||
if (type === 'abort') removeCount++
|
||||
return realRemove(type, listener, opts)
|
||||
}) as typeof ctrl.signal.removeEventListener
|
||||
|
||||
const op = jest
|
||||
.fn<() => Promise<string>>()
|
||||
.mockRejectedValueOnce(new Error('1'))
|
||||
.mockRejectedValueOnce(new Error('2'))
|
||||
.mockRejectedValueOnce(new Error('3'))
|
||||
.mockResolvedValueOnce('ok')
|
||||
|
||||
const result = await withBoundedRetry(op, {
|
||||
name: 'no-leak',
|
||||
delays: [5],
|
||||
jitter: 0,
|
||||
ttlMs: 500,
|
||||
signal: ctrl.signal
|
||||
})
|
||||
|
||||
expect(result).toBe('ok')
|
||||
expect(op).toHaveBeenCalledTimes(4)
|
||||
// If sleep() leaks listeners, addCount > removeCount. Each retry adds
|
||||
// listeners (1 sleep + 1 outer-abort forwarding) and must remove the
|
||||
// same number on settlement.
|
||||
expect(addCount).toBeGreaterThan(0)
|
||||
expect(addCount).toBe(removeCount)
|
||||
})
|
||||
|
||||
test('logger receives structured logs for retry, give-up, and recovery', async () => {
|
||||
const debugCalls: unknown[][] = []
|
||||
const infoCalls: unknown[][] = []
|
||||
const warnCalls: unknown[][] = []
|
||||
const fakeLogger = {
|
||||
debug: (...args: unknown[]) => debugCalls.push(args),
|
||||
info: (...args: unknown[]) => infoCalls.push(args),
|
||||
warn: (...args: unknown[]) => warnCalls.push(args),
|
||||
error: () => {},
|
||||
fatal: () => {},
|
||||
trace: () => {},
|
||||
child: () => fakeLogger,
|
||||
level: 'debug'
|
||||
}
|
||||
|
||||
const op = jest
|
||||
.fn<() => Promise<string>>()
|
||||
.mockRejectedValueOnce(new Error('boom'))
|
||||
.mockResolvedValueOnce('ok')
|
||||
|
||||
const result = await withBoundedRetry(op, {
|
||||
name: 'logged-op',
|
||||
delays: [5],
|
||||
jitter: 0,
|
||||
ttlMs: 200,
|
||||
logger: fakeLogger as never
|
||||
})
|
||||
|
||||
expect(result).toBe('ok')
|
||||
// Should have logged the scheduling at debug level
|
||||
expect(debugCalls.length).toBeGreaterThanOrEqual(1)
|
||||
// Should have logged the recovery at info level
|
||||
expect(infoCalls.length).toBeGreaterThanOrEqual(1)
|
||||
})
|
||||
|
||||
test('logger reports give-up via warn when TTL exceeded', async () => {
|
||||
const warnCalls: unknown[][] = []
|
||||
const fakeLogger = {
|
||||
debug: () => {},
|
||||
info: () => {},
|
||||
warn: (...args: unknown[]) => warnCalls.push(args),
|
||||
error: () => {},
|
||||
fatal: () => {},
|
||||
trace: () => {},
|
||||
child: () => fakeLogger,
|
||||
level: 'debug'
|
||||
}
|
||||
|
||||
const op = jest.fn<() => Promise<string>>().mockRejectedValue(new Error('always-fails'))
|
||||
|
||||
await expect(
|
||||
withBoundedRetry(op, {
|
||||
name: 'logged-fail',
|
||||
delays: [5],
|
||||
jitter: 0,
|
||||
ttlMs: 30,
|
||||
logger: fakeLogger as never
|
||||
})
|
||||
).rejects.toBeInstanceOf(BoundedRetryGiveUpError)
|
||||
|
||||
expect(warnCalls.length).toBeGreaterThanOrEqual(1)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,371 @@
|
||||
/**
|
||||
* Testes unitários para circuit-breaker.ts
|
||||
*/
|
||||
|
||||
import { afterEach, beforeEach, describe, expect, it } from '@jest/globals'
|
||||
import {
|
||||
CircuitBreaker,
|
||||
CircuitBreakerRegistry,
|
||||
CircuitOpenError,
|
||||
type CircuitState,
|
||||
CircuitTimeoutError,
|
||||
createCircuitBreaker,
|
||||
getCircuitHealth,
|
||||
globalCircuitRegistry,
|
||||
withCircuitBreaker
|
||||
} from '../../Utils/circuit-breaker.js'
|
||||
|
||||
describe('CircuitBreaker', () => {
|
||||
let breaker: CircuitBreaker
|
||||
|
||||
beforeEach(() => {
|
||||
breaker = createCircuitBreaker({
|
||||
name: 'test',
|
||||
failureThreshold: 3,
|
||||
successThreshold: 2,
|
||||
resetTimeout: 100,
|
||||
timeout: 1000,
|
||||
volumeThreshold: 3,
|
||||
collectMetrics: false
|
||||
})
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
breaker.destroy()
|
||||
})
|
||||
|
||||
describe('initial state', () => {
|
||||
it('should start in closed state', () => {
|
||||
expect(breaker.getState()).toBe('closed')
|
||||
expect(breaker.isClosed()).toBe(true)
|
||||
expect(breaker.isOpen()).toBe(false)
|
||||
expect(breaker.isHalfOpen()).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
describe('successful operations', () => {
|
||||
it('should execute successful operations', async () => {
|
||||
const result = await breaker.execute(() => 'success')
|
||||
expect(result).toBe('success')
|
||||
})
|
||||
|
||||
it('should execute async operations', async () => {
|
||||
const result = await breaker.execute(async () => {
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
return 'async success'
|
||||
})
|
||||
expect(result).toBe('async success')
|
||||
})
|
||||
|
||||
it('should track successful calls in stats', async () => {
|
||||
await breaker.execute(() => 'success')
|
||||
await breaker.execute(() => 'success')
|
||||
|
||||
const stats = breaker.getStats()
|
||||
expect(stats.totalCalls).toBe(2)
|
||||
expect(stats.totalSuccesses).toBe(2)
|
||||
expect(stats.totalFailures).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('failure handling', () => {
|
||||
it('should open after reaching failure threshold', async () => {
|
||||
const failingOp = () => {
|
||||
throw new Error('Failure')
|
||||
}
|
||||
|
||||
// Reach failure threshold
|
||||
for (let i = 0; i < 3; i++) {
|
||||
await expect(breaker.execute(failingOp)).rejects.toThrow('Failure')
|
||||
}
|
||||
|
||||
expect(breaker.isOpen()).toBe(true)
|
||||
})
|
||||
|
||||
it('should not open before reaching threshold', async () => {
|
||||
const failingOp = () => {
|
||||
throw new Error('Failure')
|
||||
}
|
||||
|
||||
// Below threshold
|
||||
for (let i = 0; i < 2; i++) {
|
||||
await expect(breaker.execute(failingOp)).rejects.toThrow('Failure')
|
||||
}
|
||||
|
||||
expect(breaker.isClosed()).toBe(true)
|
||||
})
|
||||
|
||||
it('should reset failure count on success', async () => {
|
||||
const failingOp = () => {
|
||||
throw new Error('Failure')
|
||||
}
|
||||
|
||||
await expect(breaker.execute(failingOp)).rejects.toThrow()
|
||||
await expect(breaker.execute(failingOp)).rejects.toThrow()
|
||||
|
||||
// Success resets failures
|
||||
await breaker.execute(() => 'success')
|
||||
|
||||
// Only 1 more failure needed (already have 2 in window, need 3 total)
|
||||
await expect(breaker.execute(failingOp)).rejects.toThrow()
|
||||
|
||||
// Should open now (3 failures in window)
|
||||
expect(breaker.isOpen()).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('open state', () => {
|
||||
beforeEach(async () => {
|
||||
// Force open
|
||||
breaker.trip()
|
||||
})
|
||||
|
||||
it('should reject operations when open', async () => {
|
||||
await expect(breaker.execute(() => 'success')).rejects.toThrow(CircuitOpenError)
|
||||
})
|
||||
|
||||
it('should transition to half-open after reset timeout', async () => {
|
||||
await new Promise(resolve => setTimeout(resolve, 150))
|
||||
expect(breaker.isHalfOpen()).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('half-open state', () => {
|
||||
beforeEach(async () => {
|
||||
breaker.trip()
|
||||
await new Promise(resolve => setTimeout(resolve, 150))
|
||||
})
|
||||
|
||||
it('should close after success threshold', async () => {
|
||||
expect(breaker.isHalfOpen()).toBe(true)
|
||||
|
||||
// Success threshold is 2
|
||||
await breaker.execute(() => 'success')
|
||||
await breaker.execute(() => 'success')
|
||||
|
||||
expect(breaker.isClosed()).toBe(true)
|
||||
})
|
||||
|
||||
it('should reopen on failure', async () => {
|
||||
expect(breaker.isHalfOpen()).toBe(true)
|
||||
|
||||
await expect(
|
||||
breaker.execute(() => {
|
||||
throw new Error('Failure')
|
||||
})
|
||||
).rejects.toThrow()
|
||||
|
||||
expect(breaker.isOpen()).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('manual controls', () => {
|
||||
it('should force open with trip()', () => {
|
||||
breaker.trip()
|
||||
expect(breaker.isOpen()).toBe(true)
|
||||
})
|
||||
|
||||
it('should force close with reset()', async () => {
|
||||
breaker.trip()
|
||||
breaker.reset()
|
||||
expect(breaker.isClosed()).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('timeout', () => {
|
||||
it('should timeout slow operations', async () => {
|
||||
const slowBreaker = createCircuitBreaker({
|
||||
name: 'slow',
|
||||
timeout: 50,
|
||||
collectMetrics: false
|
||||
})
|
||||
|
||||
await expect(slowBreaker.execute(() => new Promise(resolve => setTimeout(resolve, 200)))).rejects.toThrow(
|
||||
CircuitTimeoutError
|
||||
)
|
||||
|
||||
slowBreaker.destroy()
|
||||
})
|
||||
})
|
||||
|
||||
describe('events', () => {
|
||||
it('should emit state-change event', async () => {
|
||||
const stateChanges: Array<{ from: CircuitState; to: CircuitState }> = []
|
||||
|
||||
breaker.on('state-change', change => {
|
||||
stateChanges.push(change)
|
||||
})
|
||||
|
||||
breaker.trip()
|
||||
breaker.reset()
|
||||
|
||||
expect(stateChanges).toHaveLength(2)
|
||||
expect(stateChanges[0]).toEqual({ from: 'closed', to: 'open' })
|
||||
expect(stateChanges[1]).toEqual({ from: 'open', to: 'closed' })
|
||||
})
|
||||
|
||||
it('should emit success and failure events', async () => {
|
||||
let successCount = 0
|
||||
let failureCount = 0
|
||||
|
||||
breaker.on('success', () => successCount++)
|
||||
breaker.on('failure', () => failureCount++)
|
||||
|
||||
await breaker.execute(() => 'success')
|
||||
await expect(
|
||||
breaker.execute(() => {
|
||||
throw new Error()
|
||||
})
|
||||
).rejects.toThrow()
|
||||
|
||||
expect(successCount).toBe(1)
|
||||
expect(failureCount).toBe(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('isFailure predicate', () => {
|
||||
it('should use custom isFailure predicate', async () => {
|
||||
const customBreaker = createCircuitBreaker({
|
||||
name: 'custom',
|
||||
failureThreshold: 1,
|
||||
volumeThreshold: 1,
|
||||
shouldCountError: (error: any) => error.message !== 'Ignored',
|
||||
collectMetrics: false
|
||||
})
|
||||
|
||||
// This error should be ignored
|
||||
await expect(
|
||||
customBreaker.execute(() => {
|
||||
throw new Error('Ignored')
|
||||
})
|
||||
).rejects.toThrow()
|
||||
|
||||
expect(customBreaker.isClosed()).toBe(true)
|
||||
|
||||
// This should trip the breaker
|
||||
await expect(
|
||||
customBreaker.execute(() => {
|
||||
throw new Error('Real failure')
|
||||
})
|
||||
).rejects.toThrow()
|
||||
|
||||
expect(customBreaker.isOpen()).toBe(true)
|
||||
|
||||
customBreaker.destroy()
|
||||
})
|
||||
})
|
||||
|
||||
describe('sync operations', () => {
|
||||
it('should execute sync operations', () => {
|
||||
const result = breaker.executeSync(() => 'sync result')
|
||||
expect(result).toBe('sync result')
|
||||
})
|
||||
|
||||
it('should handle sync failures', () => {
|
||||
expect(() =>
|
||||
breaker.executeSync(() => {
|
||||
throw new Error('Sync failure')
|
||||
})
|
||||
).toThrow('Sync failure')
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('CircuitBreakerRegistry', () => {
|
||||
let registry: CircuitBreakerRegistry
|
||||
|
||||
beforeEach(() => {
|
||||
registry = new CircuitBreakerRegistry()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
registry.destroyAll()
|
||||
})
|
||||
|
||||
it('should create and retrieve circuit breakers', () => {
|
||||
const breaker1 = registry.get('test1')
|
||||
const breaker2 = registry.get('test1')
|
||||
|
||||
expect(breaker1).toBe(breaker2)
|
||||
})
|
||||
|
||||
it('should check if breaker exists', () => {
|
||||
registry.get('exists')
|
||||
|
||||
expect(registry.has('exists')).toBe(true)
|
||||
expect(registry.has('notexists')).toBe(false)
|
||||
})
|
||||
|
||||
it('should remove breaker', () => {
|
||||
registry.get('toRemove')
|
||||
expect(registry.remove('toRemove')).toBe(true)
|
||||
expect(registry.has('toRemove')).toBe(false)
|
||||
})
|
||||
|
||||
it('should get all stats', () => {
|
||||
registry.get('breaker1')
|
||||
registry.get('breaker2')
|
||||
|
||||
const stats = registry.getAllStats()
|
||||
|
||||
expect(stats).toHaveProperty('breaker1')
|
||||
expect(stats).toHaveProperty('breaker2')
|
||||
})
|
||||
|
||||
it('should reset all breakers', async () => {
|
||||
const breaker = registry.get('test')
|
||||
breaker.trip()
|
||||
|
||||
registry.resetAll()
|
||||
|
||||
expect(breaker.isClosed()).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('withCircuitBreaker', () => {
|
||||
it('should wrap function with circuit breaker', async () => {
|
||||
let callCount = 0
|
||||
|
||||
const protectedFn = withCircuitBreaker(
|
||||
async () => {
|
||||
callCount++
|
||||
return 'result'
|
||||
},
|
||||
{
|
||||
name: 'wrapped-fn',
|
||||
collectMetrics: false
|
||||
}
|
||||
)
|
||||
|
||||
const result = await protectedFn()
|
||||
|
||||
expect(result).toBe('result')
|
||||
expect(callCount).toBe(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('getCircuitHealth', () => {
|
||||
beforeEach(() => {
|
||||
globalCircuitRegistry.destroyAll()
|
||||
})
|
||||
|
||||
it('should report healthy when all circuits closed', () => {
|
||||
globalCircuitRegistry.get('healthy1')
|
||||
globalCircuitRegistry.get('healthy2')
|
||||
|
||||
const health = getCircuitHealth()
|
||||
|
||||
expect(health.healthy).toBe(true)
|
||||
expect(health.openCircuits).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('should report unhealthy when circuit is open', () => {
|
||||
const breaker = globalCircuitRegistry.get('unhealthy')
|
||||
breaker.trip()
|
||||
|
||||
const health = getCircuitHealth()
|
||||
|
||||
expect(health.healthy).toBe(false)
|
||||
expect(health.openCircuits).toContain('unhealthy')
|
||||
})
|
||||
})
|
||||
@@ -2,7 +2,7 @@
|
||||
* Testes unitários para retry-utils.ts
|
||||
*/
|
||||
|
||||
import { describe, expect, it, jest } from '@jest/globals'
|
||||
import { beforeEach, describe, expect, it, jest } from '@jest/globals'
|
||||
import {
|
||||
calculateDelay,
|
||||
createRetrier,
|
||||
@@ -11,10 +11,12 @@ import {
|
||||
retry,
|
||||
RETRY_BACKOFF_DELAYS,
|
||||
RETRY_JITTER_FACTOR,
|
||||
retryable,
|
||||
RetryAbortedError,
|
||||
retryConfigs,
|
||||
type RetryContext,
|
||||
RetryExhaustedError,
|
||||
RetryManager,
|
||||
retryPredicates,
|
||||
retryWithResult
|
||||
} from '../../Utils/retry-utils.js'
|
||||
@@ -313,6 +315,103 @@ describe('createRetrier', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('retryable', () => {
|
||||
it('should wrap function with retry', async () => {
|
||||
let attempts = 0
|
||||
|
||||
const fn = retryable(
|
||||
() => {
|
||||
attempts++
|
||||
if (attempts < 2) throw new Error('Failing')
|
||||
return 'wrapped result'
|
||||
},
|
||||
{
|
||||
maxAttempts: 5,
|
||||
baseDelay: 10,
|
||||
collectMetrics: false
|
||||
}
|
||||
)
|
||||
|
||||
const result = await fn()
|
||||
|
||||
expect(result).toBe('wrapped result')
|
||||
expect(attempts).toBe(2)
|
||||
})
|
||||
})
|
||||
|
||||
describe('RetryManager', () => {
|
||||
let manager: RetryManager
|
||||
|
||||
beforeEach(() => {
|
||||
manager = new RetryManager({ baseDelay: 200, collectMetrics: false })
|
||||
})
|
||||
|
||||
it('should execute operation with id', async () => {
|
||||
const result = await manager.execute('op1', () => 'result')
|
||||
expect(result).toBe('result')
|
||||
})
|
||||
|
||||
it('should cancel active retry', async () => {
|
||||
let attempt = 0
|
||||
|
||||
const promise = manager.execute('cancelable', async () => {
|
||||
attempt++
|
||||
if (attempt === 1) {
|
||||
throw new Error('Temporary failure')
|
||||
}
|
||||
|
||||
await new Promise(resolve => setTimeout(resolve, 1000))
|
||||
return 'result'
|
||||
})
|
||||
|
||||
// Cancel during retry delay
|
||||
setTimeout(() => manager.cancel('cancelable'), 100)
|
||||
|
||||
await expect(promise).rejects.toThrow()
|
||||
})
|
||||
|
||||
it('should check if operation is active', async () => {
|
||||
let resolve: () => void
|
||||
const promise = manager.execute(
|
||||
'active',
|
||||
() =>
|
||||
new Promise<string>(r => {
|
||||
resolve = () => r('done')
|
||||
})
|
||||
)
|
||||
|
||||
expect(manager.isActive('active')).toBe(true)
|
||||
|
||||
resolve!()
|
||||
await promise
|
||||
|
||||
expect(manager.isActive('active')).toBe(false)
|
||||
})
|
||||
|
||||
it('should cancel all operations', async () => {
|
||||
const promises = [
|
||||
manager.execute('op1', () => new Promise((_, reject) => setTimeout(() => reject(new Error()), 1000))),
|
||||
manager.execute('op2', () => new Promise((_, reject) => setTimeout(() => reject(new Error()), 1000)))
|
||||
]
|
||||
|
||||
setTimeout(() => manager.cancelAll(), 20)
|
||||
|
||||
await expect(Promise.all(promises)).rejects.toThrow()
|
||||
})
|
||||
|
||||
it('should emit events', async () => {
|
||||
const events: string[] = []
|
||||
|
||||
manager.on('attempt', () => events.push('attempt'))
|
||||
manager.on('success', () => events.push('success'))
|
||||
|
||||
await manager.execute('test', () => 'result')
|
||||
|
||||
expect(events).toContain('attempt')
|
||||
expect(events).toContain('success')
|
||||
})
|
||||
})
|
||||
|
||||
describe('retryPredicates', () => {
|
||||
describe('always', () => {
|
||||
it('should always return true', () => {
|
||||
|
||||
@@ -29,7 +29,8 @@ const TimeMs = {
|
||||
jest.mock('../../Utils/prometheus-metrics.js', () => ({
|
||||
metrics: {
|
||||
socketEvents: { inc: jest.fn() },
|
||||
errors: { inc: jest.fn() }
|
||||
errors: { inc: jest.fn() },
|
||||
circuitBreakerTrips: { inc: jest.fn() }
|
||||
}
|
||||
}))
|
||||
|
||||
@@ -58,6 +59,7 @@ describe('UnifiedSessionManager', () => {
|
||||
manager = createUnifiedSessionManager({
|
||||
enabled: true,
|
||||
logger: mockLogger as any,
|
||||
enableCircuitBreaker: false, // Disable for simpler testing
|
||||
sendNode: mockSendNode
|
||||
})
|
||||
})
|
||||
@@ -354,11 +356,7 @@ describe('shouldEnableUnifiedSession', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('UnifiedSessionManager — non-critical send semantics', () => {
|
||||
// The previous "with CircuitBreaker" suite is gone — telemetry no longer
|
||||
// uses a circuit breaker. Verify equivalent behavior still holds:
|
||||
// failures are swallowed (telemetry is non-critical) and don't propagate.
|
||||
|
||||
describe('UnifiedSessionManager with CircuitBreaker', () => {
|
||||
let manager: UnifiedSessionManager
|
||||
let mockSendNode: MockSendNode
|
||||
|
||||
@@ -368,6 +366,7 @@ describe('UnifiedSessionManager — non-critical send semantics', () => {
|
||||
manager = createUnifiedSessionManager({
|
||||
enabled: true,
|
||||
logger: createMockLogger() as any,
|
||||
enableCircuitBreaker: true,
|
||||
sendNode: mockSendNode
|
||||
})
|
||||
})
|
||||
@@ -376,13 +375,24 @@ describe('UnifiedSessionManager — non-critical send semantics', () => {
|
||||
manager.destroy()
|
||||
})
|
||||
|
||||
it('should send telemetry once per call', async () => {
|
||||
it('should work with circuit breaker enabled', async () => {
|
||||
await manager.send('login')
|
||||
expect(mockSendNode).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('should swallow send failures gracefully (telemetry is non-critical)', async () => {
|
||||
mockSendNode.mockRejectedValueOnce(new Error('boom'))
|
||||
it('should handle circuit breaker failures gracefully', async () => {
|
||||
// Make send fail multiple times to trigger circuit breaker
|
||||
let callCount = 0
|
||||
mockSendNode.mockImplementation(async () => {
|
||||
callCount++
|
||||
if (callCount <= 3) {
|
||||
throw new Error(`Fail ${callCount}`)
|
||||
}
|
||||
})
|
||||
|
||||
// These should not throw even with failures
|
||||
await expect(manager.send('login')).resolves.toBeUndefined()
|
||||
await expect(manager.send('login')).resolves.toBeUndefined()
|
||||
await expect(manager.send('login')).resolves.toBeUndefined()
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user