Doubled buckets. Part 2 (stray tcp)

.github/workflows/catch_flaky.yml

@@ -80,7 +80,7 @@ jobs:
           tt version
 
       - name: Setup luatest
-        run: tt rocks install luatest
+        run: tt rocks install luatest 1.4.1
 
       - run: cmake .
       - run: make test-flaky

.github/workflows/fast_testing.yml

@@ -107,7 +107,7 @@ jobs:
           tt version
 
       - name: Setup luatest
-        run: tt rocks install luatest
+        run: tt rocks install luatest 1.4.1
 
       - run: cmake .
       - run: make test-force

.github/workflows/reusable_testing.yml

@@ -44,7 +44,7 @@ jobs:
           tt version
 
       - name: Setup luatest
-        run: tt rocks install luatest
+        run: tt rocks install luatest 1.4.1
 
       - run: cmake .
       - run: make test-force

test/instances/storage.lua

@@ -163,7 +163,7 @@ end
 
 local function bucket_recovery_continue()
     vshard.storage.internal.errinj.ERRINJ_RECOVERY_PAUSE = false
-    vshard.storage.garbage_collector_wakeup()
+    vshard.storage.recovery_wakeup()
 end
 
 local function wal_sync()

test/luatest_helpers/vtest.lua

@@ -419,6 +419,13 @@ local function cluster_bootstrap(g, cfg)
         replicaset_count = replicaset_count + 1
     end
     t.assert_not_equals(masters, {}, 'have masters')
+    for _, master in pairs(masters) do
+        master:exec(function()
+            ilt.helpers.retrying({timeout = iwait_timeout}, function()
+                t.assert(ivshard.storage.internal.is_bucket_in_sync)
+            end)
+        end)
+    end
     vrepset.calculate_etalon_balance(etalon_balance, cfg.bucket_count)
     local fibers = table.new(0, replicaset_count)
     local bid = 1
@@ -846,6 +853,32 @@ local function info_assert_alert(alerts, alert_name)
     t.fail(('There is no %s in alerts').format(alert_name))
 end
 
+local function storage_bucket_wait_transfer(src_storage, dest_storage,
+                                            bucket_id)
+    src_storage:exec(function(bucket_id)
+        t.helpers.retrying({timeout = iwait_timeout}, function()
+            t.assert_not(box.space._bucket:get(bucket_id))
+        end)
+    end, {bucket_id})
+    dest_storage:exec(function(bucket_id)
+        t.helpers.retrying({timeout = iwait_timeout}, function()
+            t.assert_equals(box.space._bucket:get(bucket_id).status, 'active')
+        end)
+    end, {bucket_id})
+end
+
+local function storage_bucket_move(src_storage, dest_storage, bucket_id)
+    src_storage:exec(function(bucket_id, replicaset_id)
+        t.helpers.retrying({timeout = iwait_timeout}, function()
+            local res, err = ivshard.storage.bucket_send(bucket_id,
+                                                         replicaset_id)
+            t.assert_not(err)
+            t.assert(res)
+        end)
+    end, {bucket_id, dest_storage:replicaset_uuid()})
+    storage_bucket_wait_transfer(src_storage, dest_storage, bucket_id)
+end
+
 -- Git directory of the project and data directory of the test.
 -- Used in evolution tests to fetch old versions of vshard.
 local sourcedir = fio.abspath(os.getenv('PACKPACK_GIT_SOURCEDIR') or
@@ -877,6 +910,8 @@ return {
     storage_get_n_buckets = storage_get_n_buckets,
     storage_stop = storage_stop,
     storage_start = storage_start,
+    storage_bucket_move = storage_bucket_move,
+    storage_bucket_wait_transfer = storage_bucket_wait_transfer,
     router_new = router_new,
     router_cfg = router_cfg,
     router_disconnect = router_disconnect,

test/misc/check_uuid_on_connect.result

@@ -29,6 +29,10 @@ test_run:switch('bad_uuid_1_a')
 ---
 - true
 ...
+test_run:wait_log('bad_uuid_1_a', 'New master has synchronized with other replicas')
+---
+- New master has synchronized with other replicas
+...
 util = require('util')
 ---
 ...
@@ -115,10 +119,22 @@ test_run:cmd('start server bad_uuid_2_b with wait_load=False, wait=False')
 util.wait_master(test_run, REPLICASET_2, 'bad_uuid_2_a_repaired')
 ---
 ...
+test_run:switch('bad_uuid_2_a_repaired')
+---
+- true
+...
+test_run:wait_log('bad_uuid_2_a_repaired', 'New master has synchronized with other replicas')
+---
+- New master has synchronized with other replicas
+...
 test_run:switch('bad_uuid_1_a')
 ---
 - true
 ...
+test_run:wait_log('bad_uuid_1_a', 'New master has synchronized with other replicas')
+---
+- New master has synchronized with other replicas
+...
 -- Send is ok - now UUID of bad_uuid_2_a is correct.
 vshard.storage.bucket_send(1, replicaset_uuid[2])
 ---

test/misc/check_uuid_on_connect.test.lua

@@ -11,6 +11,7 @@ util.wait_master(test_run, REPLICASET_1, 'bad_uuid_1_a')
 util.wait_master(test_run, REPLICASET_2, 'bad_uuid_2_a')
 
 test_run:switch('bad_uuid_1_a')
+test_run:wait_log('bad_uuid_1_a', 'New master has synchronized with other replicas')
 util = require('util')
 vshard.storage.bucket_force_create(1)
 -- Fail, because replicaset_1 sees not the actual replicaset_2's
@@ -46,7 +47,10 @@ test_run:cmd('create server bad_uuid_2_b with script="misc/bad_uuid_2_b.lua", wa
 test_run:cmd('start server bad_uuid_2_b with wait_load=False, wait=False')
 util.wait_master(test_run, REPLICASET_2, 'bad_uuid_2_a_repaired')
 
+test_run:switch('bad_uuid_2_a_repaired')
+test_run:wait_log('bad_uuid_2_a_repaired', 'New master has synchronized with other replicas')
 test_run:switch('bad_uuid_1_a')
+test_run:wait_log('bad_uuid_1_a', 'New master has synchronized with other replicas')
 -- Send is ok - now UUID of bad_uuid_2_a is correct.
 vshard.storage.bucket_send(1, replicaset_uuid[2])
 -- Fill log with garbage to separate two 'Mismatch' messages.

test/rebalancer/bucket_ref.result

@@ -35,6 +35,10 @@ util.push_rs_filters(test_run)
 _ = test_run:switch('box_1_a')
 ---
 ...
+test_run:wait_log('box_1_a', 'New master has synchronized with other replicas')
+---
+- New master has synchronized with other replicas
+...
 vshard.storage.rebalancer_disable()
 ---
 ...
@@ -45,6 +49,10 @@ vshard.storage.bucket_force_create(1, 100)
 _ = test_run:switch('box_2_a')
 ---
 ...
+test_run:wait_log('box_2_a', 'New master has synchronized with other replicas')
+---
+- New master has synchronized with other replicas
+...
 vshard.storage.rebalancer_disable()
 ---
 ...

test/rebalancer/bucket_ref.test.lua

@@ -15,9 +15,11 @@ util.push_rs_filters(test_run)
 -- allow bucket transfer or GC during a request execution.
 --
 _ = test_run:switch('box_1_a')
+test_run:wait_log('box_1_a', 'New master has synchronized with other replicas')
 vshard.storage.rebalancer_disable()
 vshard.storage.bucket_force_create(1, 100)
 _ = test_run:switch('box_2_a')
+test_run:wait_log('box_2_a', 'New master has synchronized with other replicas')
 vshard.storage.rebalancer_disable()
 vshard.storage.bucket_force_create(101, 100)
 _ = test_run:switch('box_1_a')

test/rebalancer/errinj.result

@@ -53,6 +53,10 @@ _bucket = box.space._bucket
 vshard.storage.cfg(cfg, util.name_to_uuid.box_1_a)
 ---
 ...
+test_run:wait_log('box_1_a', 'New master has synchronized with other replicas')
+---
+- New master has synchronized with other replicas
+...
 wait_rebalancer_state('Total active bucket count is not equal to total', test_run)
 ---
 ...
@@ -67,6 +71,10 @@ test_run:switch('box_2_a')
 ---
 - true
 ...
+test_run:wait_log('box_2_a', 'New master has synchronized with other replicas')
+---
+- New master has synchronized with other replicas
+...
 _bucket = box.space._bucket
 ---
 ...
@@ -103,6 +111,10 @@ cfg.sharding[util.replicasets[1]].weight = 0.5
 vshard.storage.cfg(cfg, util.name_to_uuid.box_1_a)
 ---
 ...
+test_run:wait_log('box_1_a', 'New master has synchronized with other replicas')
+---
+- New master has synchronized with other replicas
+...
 --
 -- The rebalancer tries to send a bucket. It made the bucket be
 -- SENDING and started to persist that in WAL.

test/rebalancer/errinj.test.lua

@@ -30,11 +30,13 @@ util.push_rs_filters(test_run)
 test_run:switch('box_1_a')
 _bucket = box.space._bucket
 vshard.storage.cfg(cfg, util.name_to_uuid.box_1_a)
+test_run:wait_log('box_1_a', 'New master has synchronized with other replicas')
 wait_rebalancer_state('Total active bucket count is not equal to total', test_run)
 vshard.storage.rebalancer_disable()
 vshard.storage.bucket_force_create(1, 100)
 
 test_run:switch('box_2_a')
+test_run:wait_log('box_2_a', 'New master has synchronized with other replicas')
 _bucket = box.space._bucket
 vshard.storage.bucket_force_create(101, 100)
 vshard.storage.internal.errinj.ERRINJ_LONG_RECEIVE = true
@@ -47,6 +49,7 @@ errinj.set('ERRINJ_WAL_DELAY', true)
 log.info(string.rep('a', 1000))
 cfg.sharding[util.replicasets[1]].weight = 0.5
 vshard.storage.cfg(cfg, util.name_to_uuid.box_1_a)
+test_run:wait_log('box_1_a', 'New master has synchronized with other replicas')
 --
 -- The rebalancer tries to send a bucket. It made the bucket be
 -- SENDING and started to persist that in WAL.