@@ -566,5 +566,264 @@ describe('Watch - Unit Tests', () => {
566566 assert ( stats . cacheInvalidations > 0 ) ;
567567 } ) ;
568568 } ) ;
569+
570+ describe ( '并发安全测试 - 重连竞态条件' , ( ) => {
571+
572+ it ( 'should prevent concurrent reconnect attempts' , ( done ) => {
573+ const mockStream = new EventEmitter ( ) ;
574+ const mockCollection = {
575+ collectionName : 'test' ,
576+ watch : ( ) => new EventEmitter ( )
577+ } ;
578+ const context = {
579+ cache : { delPattern : async ( ) => 0 } ,
580+ logger : {
581+ info : ( ) => { } ,
582+ warn : ( ) => { } ,
583+ error : ( ) => { } ,
584+ debug : ( ) => { }
585+ }
586+ } ;
587+
588+ const wrapper = new ChangeStreamWrapper (
589+ mockStream ,
590+ mockCollection ,
591+ [ ] ,
592+ { reconnectInterval : 100 } ,
593+ context
594+ ) ;
595+
596+ let reconnectCount = 0 ;
597+ wrapper . on ( 'reconnect' , ( ) => {
598+ reconnectCount ++ ;
599+ } ) ;
600+
601+ // 模拟多个并发错误
602+ const error1 = new Error ( 'Connection reset' ) ;
603+ error1 . code = 'ECONNRESET' ;
604+
605+ const error2 = new Error ( 'Connection timeout' ) ;
606+ error2 . code = 'ETIMEDOUT' ;
607+
608+ // 同时触发多个错误
609+ mockStream . emit ( 'error' , error1 ) ;
610+ mockStream . emit ( 'error' , error2 ) ;
611+ mockStream . emit ( 'error' , error1 ) ;
612+
613+ // 等待重连尝试
614+ setTimeout ( ( ) => {
615+ // 应该只有一次重连尝试（第二和第三次被阻止）
616+ assert . strictEqual ( reconnectCount , 1 , 'Should only have 1 reconnect attempt' ) ;
617+ wrapper . close ( ) ;
618+ done ( ) ;
619+ } , 200 ) ;
620+ } ) ;
621+
622+ it ( 'should handle reconnect failure without infinite recursion' , ( done ) => {
623+ let watchCallCount = 0 ;
624+ const mockStream = new EventEmitter ( ) ;
625+ const mockCollection = {
626+ collectionName : 'test' ,
627+ watch : ( ) => {
628+ watchCallCount ++ ;
629+ // 模拟 watch 总是失败
630+ throw new Error ( 'Watch failed' ) ;
631+ }
632+ } ;
633+ const context = {
634+ cache : { delPattern : async ( ) => 0 } ,
635+ logger : {
636+ info : ( ) => { } ,
637+ warn : ( ) => { } ,
638+ error : ( ) => { } ,
639+ debug : ( ) => { }
640+ }
641+ } ;
642+
643+ const wrapper = new ChangeStreamWrapper (
644+ mockStream ,
645+ mockCollection ,
646+ [ ] ,
647+ { reconnectInterval : 50 , maxReconnectDelay : 100 } ,
648+ context
649+ ) ;
650+
651+ // 触发重连
652+ const error = new Error ( 'Connection reset' ) ;
653+ error . code = 'ECONNRESET' ;
654+ mockStream . emit ( 'error' , error ) ;
655+
656+ // 等待足够长时间观察是否有无限递归
657+ setTimeout ( ( ) => {
658+ // 应该有重连尝试，但不应该有大量调用（说明没有无限递归）
659+ assert ( watchCallCount > 0 , 'Should have reconnect attempts' ) ;
660+ assert ( watchCallCount < 10 , 'Should not have excessive reconnect attempts' ) ;
661+ wrapper . close ( ) ;
662+ done ( ) ;
663+ } , 500 ) ;
664+ } ) ;
665+
666+ it ( 'should reset reconnecting flag even on error' , ( done ) => {
667+ const mockStream = new EventEmitter ( ) ;
668+ let watchCalls = 0 ;
669+ const mockCollection = {
670+ collectionName : 'test' ,
671+ watch : ( ) => {
672+ watchCalls ++ ;
673+ if ( watchCalls === 1 ) {
674+ // 第一次失败
675+ throw new Error ( 'Watch failed' ) ;
676+ } else {
677+ // 第二次成功
678+ return new EventEmitter ( ) ;
679+ }
680+ }
681+ } ;
682+ const context = {
683+ cache : { delPattern : async ( ) => 0 } ,
684+ logger : {
685+ info : ( ) => { } ,
686+ warn : ( ) => { } ,
687+ error : ( ) => { } ,
688+ debug : ( ) => { }
689+ }
690+ } ;
691+
692+ const wrapper = new ChangeStreamWrapper (
693+ mockStream ,
694+ mockCollection ,
695+ [ ] ,
696+ { reconnectInterval : 50 } ,
697+ context
698+ ) ;
699+
700+ // 触发第一次重连
701+ const error = new Error ( 'Connection reset' ) ;
702+ error . code = 'ECONNRESET' ;
703+ mockStream . emit ( 'error' , error ) ;
704+
705+ // 等待第一次重连完成并失败
706+ setTimeout ( ( ) => {
707+ // 检查已经有重连尝试
708+ const initialAttempts = wrapper . getStats ( ) . reconnectAttempts ;
709+ assert ( initialAttempts > 0 , 'Should have initial reconnect attempt' ) ;
710+
711+ // 触发第二次重连（应该能成功，说明标志已重置）
712+ mockStream . emit ( 'error' , error ) ;
713+
714+ setTimeout ( ( ) => {
715+ // 检查有第二次重连尝试
716+ assert ( watchCalls >= 2 , `Should have multiple watch calls, got ${ watchCalls } ) ;
717+ wrapper . close ( ) ;
718+ done ( ) ;
719+ } , 300 ) ;
720+ } , 250 ) ;
721+ } ) ;
722+ } ) ;
723+
724+ describe ( '错误处理递归防护' , ( ) => {
725+
726+ it ( 'should use setTimeout to prevent synchronous recursion' , ( done ) => {
727+ const mockStream = new EventEmitter ( ) ;
728+ let recursionDepth = 0 ;
729+ let maxRecursionDepth = 0 ;
730+
731+ const mockCollection = {
732+ collectionName : 'test' ,
733+ watch : ( ) => {
734+ recursionDepth ++ ;
735+ maxRecursionDepth = Math . max ( maxRecursionDepth , recursionDepth ) ;
736+
737+ // 模拟失败
738+ const failedStream = new EventEmitter ( ) ;
739+ setTimeout ( ( ) => {
740+ const error = new Error ( 'Watch creation failed' ) ;
741+ failedStream . emit ( 'error' , error ) ;
742+ recursionDepth -- ;
743+ } , 10 ) ;
744+
745+ return failedStream ;
746+ }
747+ } ;
748+
749+ const context = {
750+ cache : { delPattern : async ( ) => 0 } ,
751+ logger : {
752+ info : ( ) => { } ,
753+ warn : ( ) => { } ,
754+ error : ( ) => { } ,
755+ debug : ( ) => { }
756+ }
757+ } ;
758+
759+ const wrapper = new ChangeStreamWrapper (
760+ mockStream ,
761+ mockCollection ,
762+ [ ] ,
763+ { reconnectInterval : 20 } ,
764+ context
765+ ) ;
766+
767+ // 触发初始错误
768+ const error = new Error ( 'Initial error' ) ;
769+ error . code = 'ECONNRESET' ;
770+ mockStream . emit ( 'error' , error ) ;
771+
772+ // 等待观察递归深度
773+ setTimeout ( ( ) => {
774+ // 递归深度应该很浅（因为使用了 setTimeout 避免同步递归）
775+ assert ( maxRecursionDepth <= 2 , `Max recursion depth should be <= 2, got ${ maxRecursionDepth } ) ;
776+ wrapper . close ( ) ;
777+ done ( ) ;
778+ } , 300 ) ;
779+ } ) ;
780+
781+ it ( 'should check closed flag before retry' , ( done ) => {
782+ const mockStream = new EventEmitter ( ) ;
783+ const mockCollection = {
784+ collectionName : 'test' ,
785+ watch : ( ) => {
786+ throw new Error ( 'Watch failed' ) ;
787+ }
788+ } ;
789+
790+ let errorHandlerCalled = 0 ;
791+ const context = {
792+ cache : { delPattern : async ( ) => 0 } ,
793+ logger : {
794+ info : ( ) => { } ,
795+ warn : ( ) => { } ,
796+ error : ( ) => { errorHandlerCalled ++ ; } ,
797+ debug : ( ) => { }
798+ }
799+ } ;
800+
801+ const wrapper = new ChangeStreamWrapper (
802+ mockStream ,
803+ mockCollection ,
804+ [ ] ,
805+ { reconnectInterval : 50 } ,
806+ context
807+ ) ;
808+
809+ // 触发错误
810+ const error = new Error ( 'Connection reset' ) ;
811+ error . code = 'ECONNRESET' ;
812+ mockStream . emit ( 'error' , error ) ;
813+
814+ // 立即关闭 wrapper
815+ setTimeout ( ( ) => {
816+ wrapper . close ( ) ;
817+ } , 20 ) ;
818+
819+ // 等待确认不会继续重试
820+ setTimeout ( ( ) => {
821+ const attempts = wrapper . getStats ( ) . reconnectAttempts ;
822+ // 应该只有很少的重试（因为已关闭）
823+ assert ( attempts <= 2 , `Should have minimal retries after close, got ${ attempts } ) ;
824+ done ( ) ;
825+ } , 300 ) ;
826+ } ) ;
827+ } ) ;
569828} ) ;
570829
0 commit comments