@@ -541,6 +541,217 @@ func TestGetRemovalMultiChannel(t *testing.T) {
541541 require .Equal (t , bodyExpected , body )
542542}
543543
544+ // TestDeltaSyncConcurrentClientCachePopulation tests delta sync behavior when multiple goroutines request the same delta simultaneously.
545+ // Ensures that only one delta is computed and others wait for the cached result.
546+ // More instances of duplicated delta computation will be seen for larger documents since the delta computation takes longer.
547+ func TestDeltaSyncConcurrentClientCachePopulation (t * testing.T ) {
548+ if ! base .IsEnterpriseEdition () {
549+ t .Skip ("Delta sync only supported in EE" )
550+ }
551+
552+ tests := []struct {
553+ name string
554+ docSize int
555+ concurrentClients int
556+ }{
557+ {
558+ name : "100KBDoc_100Clients" ,
559+ docSize : 100 * 1024 ,
560+ concurrentClients : 100 ,
561+ },
562+ {
563+ name : "100KBDoc_1000Clients" ,
564+ docSize : 100 * 1024 ,
565+ concurrentClients : 1000 ,
566+ },
567+ {
568+ name : "100KBDoc_5000Clients" ,
569+ docSize : 100 * 1024 ,
570+ concurrentClients : 5000 ,
571+ },
572+ {
573+ name : "5MBDoc_10Clients" ,
574+ docSize : 5 * 1024 * 1024 ,
575+ concurrentClients : 10 ,
576+ },
577+ {
578+ name : "5MBDoc_100Clients" ,
579+ docSize : 5 * 1024 * 1024 ,
580+ concurrentClients : 100 ,
581+ },
582+ {
583+ name : "5MBDoc_1000Clients" ,
584+ docSize : 5 * 1024 * 1024 ,
585+ concurrentClients : 1000 ,
586+ },
587+ {
588+ name : "5MBDoc_5000Clients" ,
589+ docSize : 5 * 1024 * 1024 ,
590+ concurrentClients : 5000 ,
591+ },
592+ }
593+
594+ for _ , test := range tests {
595+ t .Run (test .name , func (t * testing.T ) {
596+ db , ctx := SetupTestDBWithOptions (t , DatabaseContextOptions {DeltaSyncOptions : DeltaSyncOptions {Enabled : true , RevMaxAgeSeconds : 300 }})
597+ defer db .Close (ctx )
598+ collection , ctx := GetSingleDatabaseCollectionWithUser (ctx , t , db )
599+
600+ docID := "doc1_" + test .name
601+ rev1 , _ , err := collection .Put (ctx , docID , Body {"foo" : "bar" , "bar" : "buzz" , "quux" : strings .Repeat ("a" , test .docSize )})
602+ require .NoError (t , err )
603+ rev2 , _ , err := collection .Put (ctx , docID , Body {"foo" : "bar" , "quux" : strings .Repeat ("b" , test .docSize ), BodyRev : rev1 })
604+ require .NoError (t , err )
605+
606+ wg := sync.WaitGroup {}
607+ wg .Add (test .concurrentClients )
608+ for i := 0 ; i < test .concurrentClients ; i ++ {
609+ go func () {
610+ defer wg .Done ()
611+ delta , _ , err := collection .GetDelta (ctx , docID , rev1 , rev2 )
612+ require .NoErrorf (t , err , "Error getting delta for doc %q from rev %q to %q" , docID , rev1 , rev2 )
613+ require .NotNil (t , delta )
614+ }()
615+ }
616+ wg .Wait ()
617+
618+ // ensure only 1 delta miss and the remaining used cache
619+ deltaCacheHits := db .DbStats .DeltaSync ().DeltaCacheHit .Value ()
620+ deltaCacheMisses := db .DbStats .DeltaSync ().DeltaCacheMiss .Value ()
621+ assert .Equal (t , int64 (1 ), deltaCacheMisses , "Unexpected number of delta cache misses" )
622+ assert .Equal (t , int64 (test .concurrentClients - 1 ), deltaCacheHits , "Unexpected number of delta cache hits" )
623+ })
624+ }
625+ }
626+
627+ func BenchmarkDeltaSyncConcurrentClientCachePopulation (b * testing.B ) {
628+ if ! base .IsEnterpriseEdition () {
629+ b .Skip ("Delta sync only supported in EE" )
630+ }
631+
632+ tests := []struct {
633+ name string
634+ docSize int
635+ concurrentClients int
636+ }{
637+ {
638+ name : "100KBDoc_1Client" ,
639+ docSize : 100 * 1024 ,
640+ concurrentClients : 1 ,
641+ },
642+ {
643+ name : "100KBDoc_100Clients" ,
644+ docSize : 100 * 1024 ,
645+ concurrentClients : 100 ,
646+ },
647+ {
648+ name : "100KBDoc_1000Clients" ,
649+ docSize : 100 * 1024 ,
650+ concurrentClients : 1000 ,
651+ },
652+ {
653+ name : "100KBDoc_5000Clients" ,
654+ docSize : 100 * 1024 ,
655+ concurrentClients : 5000 ,
656+ },
657+ {
658+ name : "5MBDoc_1Client" ,
659+ docSize : 5 * 1024 * 1024 ,
660+ concurrentClients : 1 ,
661+ },
662+ {
663+ name : "5MBDoc_10Clients" ,
664+ docSize : 5 * 1024 * 1024 ,
665+ concurrentClients : 10 ,
666+ },
667+ {
668+ name : "5MBDoc_100Clients" ,
669+ docSize : 5 * 1024 * 1024 ,
670+ concurrentClients : 100 ,
671+ },
672+ {
673+ name : "5MBDoc_1000Clients" ,
674+ docSize : 5 * 1024 * 1024 ,
675+ concurrentClients : 1000 ,
676+ },
677+ {
678+ name : "5MBDoc_5000Clients" ,
679+ docSize : 5 * 1024 * 1024 ,
680+ concurrentClients : 5000 ,
681+ },
682+ }
683+
684+ for _ , test := range tests {
685+ b .Run (test .name , func (b * testing.B ) {
686+ db , ctx := SetupTestDBWithOptions (b , DatabaseContextOptions {DeltaSyncOptions : DeltaSyncOptions {Enabled : true , RevMaxAgeSeconds : 300 }})
687+ defer db .Close (ctx )
688+ collection , ctx := GetSingleDatabaseCollectionWithUser (ctx , b , db )
689+
690+ docID := "doc1_" + test .name
691+ rev1 , _ , _ := collection .Put (ctx , docID , Body {"foo" : "bar" , "bar" : "buzz" , "quux" : strings .Repeat ("a" , test .docSize )})
692+ rev2 , _ , _ := collection .Put (ctx , docID , Body {"foo" : "bar" , "quux" : strings .Repeat ("b" , test .docSize ), BodyRev : rev1 })
693+
694+ for b .Loop () {
695+ wg := sync.WaitGroup {}
696+ wg .Add (test .concurrentClients )
697+ for i := 0 ; i < test .concurrentClients ; i ++ {
698+ go func () {
699+ defer wg .Done ()
700+ _ , _ , _ = collection .GetDelta (ctx , docID , rev1 , rev2 )
701+ }()
702+ }
703+ wg .Wait ()
704+ }
705+ })
706+ }
707+ }
708+
709+ func BenchmarkDeltaSyncSingleClientCachePopulation (b * testing.B ) {
710+ if ! base .IsEnterpriseEdition () {
711+ b .Skip ("Delta sync only supported in EE" )
712+ }
713+
714+ tests := []struct {
715+ name string
716+ docSize int
717+ }{
718+ {
719+ name : "100KBDoc" ,
720+ docSize : 100 * 1024 ,
721+ },
722+ //{
723+ // name: "5MBDoc",
724+ // docSize: 5 * 1024 * 1024,
725+ //},
726+ }
727+
728+ for _ , test := range tests {
729+ b .Run (test .name , func (b * testing.B ) {
730+ db , ctx := SetupTestDBWithOptions (b , DatabaseContextOptions {DeltaSyncOptions : DeltaSyncOptions {Enabled : true , RevMaxAgeSeconds : 300 }})
731+ defer db .Close (ctx )
732+ collection , ctx := GetSingleDatabaseCollectionWithUser (ctx , b , db )
733+
734+ // run benchmark over 1000 non-cached deltas since repeated invocations of b.Loop would hit the cache
735+ const numDocs = 1000
736+ benchDocs := make ([]struct { docID , rev1 , rev2 string }, 0 , numDocs )
737+ for i := range numDocs {
738+ docID := fmt .Sprintf ("%s_doc_%d" , test .name , i )
739+ rev1 , _ , err := collection .Put (ctx , docID , Body {"foo" : "bar" , "bar" : "buzz" , "quux" : strings .Repeat ("a" , test .docSize )})
740+ require .NoError (b , err )
741+ rev2 , _ , err := collection .Put (ctx , docID , Body {"foo" : "bar" , "quux" : strings .Repeat ("b" , test .docSize ), BodyRev : rev1 })
742+ require .NoError (b , err )
743+ benchDocs = append (benchDocs , struct { docID , rev1 , rev2 string }{docID : docID , rev1 : rev1 , rev2 : rev2 })
744+ }
745+
746+ for b .Loop () {
747+ for _ , doc := range benchDocs {
748+ _ , _ , _ = collection .GetDelta (ctx , doc .docID , doc .rev1 , doc .rev2 )
749+ }
750+ }
751+ })
752+ }
753+ }
754+
544755// Test delta sync behavior when the fromRevision is a channel removal.
545756func TestDeltaSyncWhenFromRevIsChannelRemoval (t * testing.T ) {
546757 db , ctx := setupTestDB (t )
0 commit comments