test_run = require('test_run').new()
REPLICASET_1 = { 'storage_1_a', 'storage_1_b' }
REPLICASET_2 = { 'storage_2_a', 'storage_2_b' }

test_run:create_cluster(REPLICASET_1, 'storage')
test_run:create_cluster(REPLICASET_2, 'storage')
util = require('util')
util.wait_master(test_run, REPLICASET_1, 'storage_1_a')
util.wait_master(test_run, REPLICASET_2, 'storage_2_a')
util.push_rs_filters(test_run)

_ = test_run:switch("storage_2_a")
-- Pause until restart. Otherwise recovery does its job too fast and does not
-- allow to simulate the intermediate state.
vshard.storage.internal.errinj.ERRINJ_RECOVERY_PAUSE = true
vshard.storage.rebalancer_disable()

_ = test_run:switch('default')
util.map_bucket_protection(test_run, {REPLICASET_1}, false)

_ = test_run:switch('storage_1_a')
vshard.storage.internal.errinj.ERRINJ_RECOVERY_PAUSE = true

-- Create buckets sending to rs2 and restart - recovery must
-- garbage some of them and activate others. Receiving buckets
-- must be garbaged on bootstrap.
_bucket = box.space._bucket
_bucket:replace{2, vshard.consts.BUCKET.SENDING, util.replicasets[2]}
_bucket:replace{3, vshard.consts.BUCKET.RECEIVING, util.replicasets[2]}
vshard.storage.sync()

_ = test_run:switch('default')
util.map_bucket_protection(test_run, {REPLICASET_1}, true)
util.map_bucket_protection(test_run, {REPLICASET_2}, false)

_ = test_run:switch('storage_2_a')
_bucket = box.space._bucket
vshard.storage.bucket_force_create(2)
_bucket:replace{3, vshard.consts.BUCKET.SENDING, util.replicasets[1]}
vshard.storage.sync()

_ = test_run:switch('default')
util.map_bucket_protection(test_run, {REPLICASET_2}, true)

_ = test_run:cmd('stop server storage_1_a')
_ = test_run:cmd('start server storage_1_a')

_ = test_run:switch('storage_1_a')
_bucket = box.space._bucket
while _bucket:count() ~= 0 do vshard.storage.recovery_wakeup() fiber.sleep(0.01) end

_ = test_run:switch('storage_2_a')
_bucket:select{}

_ = test_run:switch('storage_1_a')

--
-- Test a case, when a bucket is sending on one replicaset,
-- receiving on another one, but there is no rebalancing.
--
_ = test_run:cmd('stop server storage_2_a')
_ = test_run:cmd('start server storage_2_a')
_ = test_run:switch('storage_2_a')
_bucket = box.space._bucket
while _bucket:count() ~= 2 do vshard.storage.recovery_wakeup() fiber.sleep(0.1) end

--
-- Test a case, when a destination is down. The recovery fiber
-- must restore buckets, when the destination is up.
--
_ = test_run:switch('default')
util.map_bucket_protection(test_run, {REPLICASET_1}, false)

_ = test_run:switch('storage_1_a')
vshard.storage.internal.errinj.ERRINJ_RECOVERY_PAUSE = true
_bucket:replace{1, vshard.consts.BUCKET.SENDING, util.replicasets[2]}
vshard.storage.sync()
_ = test_run:switch('storage_2_a')
vshard.storage.bucket_force_create(1)
vshard.storage.sync()

_ = test_run:switch('default')
util.map_bucket_protection(test_run, {REPLICASET_1}, true)
_ = test_run:cmd('stop server storage_2_a')
_ = test_run:cmd('stop server storage_1_a')
_ = test_run:cmd('start server storage_1_a')
_ = test_run:switch('storage_1_a')
_bucket = box.space._bucket
_bucket:select{}
for i = 1, 10 do vshard.storage.recovery_wakeup() end
_bucket:select{}
_ = test_run:cmd('start server storage_2_a')
while _bucket:count() ~= 0 do vshard.storage.recovery_wakeup() fiber.sleep(0.1) end
_bucket:select{}

_ = test_run:switch('storage_2_a')
_bucket = box.space._bucket
_bucket:select{}

--
-- Test a case when a bucket is sending in one place and garbage
-- or sent or deleted on a destination.
--
_ = test_run:switch('default')
util.map_bucket_protection(test_run, {REPLICASET_1, REPLICASET_2}, false)

_ = test_run:switch('storage_2_a')
_bucket:replace{1, vshard.consts.BUCKET.GARBAGE, util.replicasets[1]}
vshard.storage.sync()
_ = test_run:switch('storage_1_a')
_bucket:replace{1, vshard.consts.BUCKET.SENDING, util.replicasets[2]}
vshard.storage.sync()

_ = test_run:switch('default')
util.map_bucket_protection(test_run, {REPLICASET_1, REPLICASET_2}, true)

_ = test_run:switch('default')
_ = test_run:cmd('stop server storage_2_a')
_ = test_run:cmd('stop server storage_1_a')
_ = test_run:cmd('start server storage_1_a')
_ = test_run:cmd('start server storage_2_a')
_ = test_run:switch('storage_1_a')
_bucket = box.space._bucket
while _bucket:get{1}.status ~= vshard.consts.BUCKET.ACTIVE do vshard.storage.recovery_wakeup() fiber.sleep(0.1) end

_ = test_run:switch("default")
test_run:drop_cluster(REPLICASET_2)
test_run:drop_cluster(REPLICASET_1)
_ = test_run:cmd('clear filter')
