@@ -604,4 +604,144 @@ describe('Dealer', () => {
604604
605605 await runDealer ( dealer ) ;
606606 } ) ;
607+
608+ describe ( 'setLimit' , ( ) => {
609+ test ( 'increasing the limit immediately fills newly available slots' , async ( ) => {
610+ const items = createItems ( 4 ) ;
611+ let maxConcurrent = 0 ;
612+ let currentConcurrent = 0 ;
613+ const dealer = new Dealer ( items , { limit : 1 } ) ;
614+ const { promise : firstStarted , resolve : resolveFirstStarted } =
615+ Promise . withResolvers < void > ( ) ;
616+ const { promise : canFinishFirst , resolve : resolveCanFinishFirst } =
617+ Promise . withResolvers < void > ( ) ;
618+ let firstCall = true ;
619+
620+ await dealer . setup ( ( ) => {
621+ return Promise . resolve ( async ( ) => {
622+ currentConcurrent ++ ;
623+ maxConcurrent = Math . max ( maxConcurrent , currentConcurrent ) ;
624+ if ( firstCall ) {
625+ firstCall = false ;
626+ resolveFirstStarted ( ) ;
627+ await canFinishFirst ;
628+ }
629+ currentConcurrent -- ;
630+ } ) ;
631+ } ) ;
632+
633+ const done = runDealer ( dealer ) ;
634+ await firstStarted ;
635+ // limit 1 のあいだは1件しか動いていないはず
636+ expect ( maxConcurrent ) . toBe ( 1 ) ;
637+
638+ dealer . setLimit ( 4 ) ;
639+ resolveCanFinishFirst ( ) ;
640+ await done ;
641+
642+ expect ( maxConcurrent ) . toBeGreaterThan ( 1 ) ;
643+ expect ( dealer . limit ) . toBe ( 4 ) ;
644+ } ) ;
645+
646+ test ( 'decreasing the limit lets already-started workers finish, but throttles concurrency for items dispatched afterward' , async ( ) => {
647+ // limit 3・5件: 最初の3件（index 0,1,2）が同時ディスパッチされる。
648+ // それらが実行中のうちに limit を 1 へ落とし、(a) 実行中の3件は
649+ // 中断されず全件完了すること、(b) 減少後に新規ディスパッチされる
650+ // 残り2件（index 3,4）は同時に1件までしか動かないことを検証する。
651+ const items = createItems ( 5 ) ;
652+ const dealer = new Dealer ( items , { limit : 3 } ) ;
653+ const startedFirstBatch : number [ ] = [ ] ;
654+ const { promise : firstBatchStarted , resolve : resolveFirstBatchStarted } =
655+ Promise . withResolvers < void > ( ) ;
656+ const { promise : canFinishFirstBatch , resolve : resolveCanFinishFirstBatch } =
657+ Promise . withResolvers < void > ( ) ;
658+ let decreased = false ;
659+ let concurrentAfterDecrease = 0 ;
660+ let maxConcurrentAfterDecrease = 0 ;
661+
662+ await dealer . setup ( ( _item , index ) => {
663+ return Promise . resolve ( async ( ) => {
664+ if ( ! decreased ) {
665+ startedFirstBatch . push ( index ) ;
666+ if ( startedFirstBatch . length === 3 ) {
667+ resolveFirstBatchStarted ( ) ;
668+ }
669+ await canFinishFirstBatch ;
670+ return ;
671+ }
672+ concurrentAfterDecrease ++ ;
673+ maxConcurrentAfterDecrease = Math . max (
674+ maxConcurrentAfterDecrease ,
675+ concurrentAfterDecrease ,
676+ ) ;
677+ await new Promise ( ( r ) => setTimeout ( r , 5 ) ) ;
678+ concurrentAfterDecrease -- ;
679+ } ) ;
680+ } ) ;
681+
682+ const done = runDealer ( dealer ) ;
683+ await firstBatchStarted ;
684+ expect ( startedFirstBatch ) . toHaveLength ( 3 ) ;
685+
686+ // setLimit はテストの非同期フロー（＝ワーカー自身の同期区間の外）から
687+ // 呼ぶ。これは実運用（外部入力による並列数変更）と同じ呼び出し方。
688+ decreased = true ;
689+ dealer . setLimit ( 1 ) ;
690+ resolveCanFinishFirstBatch ( ) ;
691+
692+ await done ;
693+
694+ expect ( startedFirstBatch ) . toHaveLength ( 3 ) ;
695+ expect ( maxConcurrentAfterDecrease ) . toBe ( 1 ) ;
696+ expect ( dealer . limit ) . toBe ( 1 ) ;
697+ } ) ;
698+
699+ test ( 'throws RangeError for non-positive-integer limits' , ( ) => {
700+ const dealer = new Dealer ( createItems ( 1 ) , { limit : 5 } ) ;
701+ expect ( ( ) => dealer . setLimit ( 0 ) ) . toThrow ( RangeError ) ;
702+ expect ( ( ) => dealer . setLimit ( - 1 ) ) . toThrow ( RangeError ) ;
703+ expect ( ( ) => dealer . setLimit ( 1.5 ) ) . toThrow ( RangeError ) ;
704+ expect ( dealer . limit ) . toBe ( 5 ) ;
705+ } ) ;
706+
707+ test ( 'after the dealer has finished, setLimit updates the stored limit without throwing or dispatching' , async ( ) => {
708+ const items = createItems ( 1 ) ;
709+ const dealer = new Dealer ( items , { limit : 10 } ) ;
710+
711+ await dealer . setup ( ( ) => Promise . resolve ( ( ) => { } ) ) ;
712+ await runDealer ( dealer ) ;
713+
714+ expect ( ( ) => dealer . setLimit ( 3 ) ) . not . toThrow ( ) ;
715+ // #deal() 自体は #finished ガードで即 return する（新規ディスパッチは
716+ // 発生しない）が、#limit フィールドの更新はガードの影響を受けない
717+ expect ( dealer . limit ) . toBe ( 3 ) ;
718+ } ) ;
719+
720+ test ( 'a synchronous setLimit call from within a worker still respects the limit deterministically (re-entrant #deal() calls are ignored)' , async ( ) => {
721+ // worker 自身の同期区間（最初の await より前）から setLimit を呼ぶ
722+ // 稀なケースでも、#deal() の再入防止により外側のディスパッチループが
723+ // 一貫して最新の #limit を尊重する。increase 版は
724+ // 'onStart receives a controller...'（deal.spec.ts）で間接的に検証済み
725+ // なので、ここでは decrease 版のみ確認する。
726+ const items = createItems ( 3 ) ;
727+ const dealer = new Dealer ( items , { limit : 3 } ) ;
728+ const processed : number [ ] = [ ] ;
729+
730+ await dealer . setup ( ( _item , index ) => {
731+ return Promise . resolve ( async ( ) => {
732+ processed . push ( index ) ;
733+ if ( index === 0 ) {
734+ // この時点で #deal() はまだ while ループの最中（再入）
735+ dealer . setLimit ( 1 ) ;
736+ }
737+ await new Promise ( ( r ) => setTimeout ( r , 1 ) ) ;
738+ } ) ;
739+ } ) ;
740+
741+ await runDealer ( dealer ) ;
742+ // 再入経路でも例外や取りこぼしなく全件完了する
743+ expect ( processed ) . toHaveLength ( 3 ) ;
744+ expect ( dealer . limit ) . toBe ( 1 ) ;
745+ } ) ;
746+ } ) ;
607747} ) ;
0 commit comments