MV update may crash during bootstrap

[gleb-cloudius]

Reproducer:

diff --git a/test/topology_custom/test_mv_decommission.py b/test/topology_custom/test_mv_decommission.py
new file mode 100644
index 0000000000..caff758b71
--- /dev/null
+++ b/test/topology_custom/test_mv_decommission.py
@@ -0,0 +1,74 @@
+import asyncio
+import pytest
+import time
+import logging
+
+from cassandra.cluster import ConnectionException, ConsistencyLevel, NoHostAvailable, Session, SimpleStatement  # type: ignore
+
+from test.pylib.manager_client import ManagerClient
+from test.topology.conftest import skip_mode
+
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.asyncio
+@skip_mode('release', 'error injections are not supported in release mode')
+async def test_mv_decommission(manager: ManagerClient):
+    cfg = {'force_gossip_topology_changes': True}
+
+    servers = [await manager.server_add(config=cfg, timeout=60) for _ in range(3)]
+
+    cql = manager.get_cql()
+    await cql.run_async("CREATE KEYSPACE ks WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 3};")
+    await cql.run_async("CREATE TABLE ks.t (pk int primary key, v int)")
+    await cql.run_async("CREATE materialized view ks.t_view AS select pk, v from ks.t where v is not null primary key (v, pk)")
+
+    stop_event = asyncio.Event()
+    concurrency = 10
+    async def do_writes() -> int:
+        iteration = 0
+        while not stop_event.is_set():
+            start_time = time.time()
+            try:
+                await cql.run_async(f"insert into ks.t (pk, v) values ({iteration}, {iteration})")
+            except NoHostAvailable as e:
+                for _, err in e.errors.items():
+                    # ConnectionException can be raised when the node is shutting down.
+                    if not isinstance(err, ConnectionException):
+                        logger.error(f"Write started {time.time() - start_time}s ago failed: {e}")
+                        raise
+            except Exception as e:
+                logger.error(f"Write started {time.time() - start_time}s ago failed: {e}")
+                raise
+            iteration += 1
+            await asyncio.sleep(0.01)
+
+        return iteration
+
+    tasks = [asyncio.create_task(do_writes()) for _ in range(concurrency)]
+
+    async def finish() -> list[int]:
+        logger.info("Stopping write workers")
+        stop_event.set()
+        nums = await asyncio.gather(*tasks)
+        return nums
+
+    await manager.server_add()
+    await manager.server_add()
+    await manager.server_add()
+
+    nums = await finish()
+
+    await asyncio.sleep(2)
+
+    logger.info(nums)
+    logger.info(sum(nums))
+
+    res = await cql.run_async("select pk, v from ks.t");
+    for r in res:
+        logger.info(f"pk: {r.pk}, v: {r.v}")
+
+    res = await cql.run_async("select v, pk from ks.t_view");
+    for r in res:
+        logger.info(f"v: {r.v}, pk: {r.pk}")
diff --git a/db/view/view.cc b/db/view/view.cc
index 8807a6748d..6a2d419ad8 100644
--- a/db/view/view.cc
+++ b/db/view/view.cc
@@ -1714,7 +1714,8 @@ future<> view_update_generator::mutate_MV(
         wait_for_all_updates wait_for_all)
 {
     auto base_ermp = base->table().get_effective_replication_map();
-    static constexpr size_t max_concurrent_updates = 128;
+    static constexpr size_t max_concurrent_updates = 2;
+    co_await seastar::sleep(std::chrono::milliseconds{1000});
     co_await max_concurrent_for_each(view_updates, max_concurrent_updates,
             [this, base_token, &stats, &cf_stats, tr_state, &pending_view_updates, allow_hints, wait_for_all, base_ermp] (frozen_mutation_and_schema mut) mutable -> future<> {
         auto view_token = dht::get_token(*mut.s, mut.fm.key());

The reason is that taking of a base table and view tables erm is not atomic (max_concurrent_for_each may preempt). The code in get_view_natural_endpoint though assumes that typologies are the same since it has code like:

            } else if (!network_topology || topology.get_datacenter(view_endpoint) == my_datacenter) {
                view_endpoints.push_back(view_endpoint);
            }

Where it looks up DC for view endpoint in base table topology. This may fail if view topology contain new node, while base table does not.

[gleb-cloudius]

Similar issue: #17786

[kbr-scylla]

The reason is that taking of a base table and view tables erm is not atomic (max_concurrent_for_each may preempt). The code in get_view_natural_endpoint though assumes that typologies are the same since it has code like:

So it's a bug in MV code. But seems easy to fix, e.g. by ensuring we don't yield between taking the erms, or perhaps by taking the locations from the same erm? (locations for base endpoints from view erm?)

Anyway since it's MV code bug, reassigning to @nyh (although not sure, maybe it should be @piodul with the recent changes cc @kostja? Dunno -- please reassign if necessary)

Also, since this looks pretty serious -- it could potentially cause a full cluster crash when we have MV workload during bootstrap -- marking as release blocker

[mykaul]

@kostja , @piodul - who's handling it?

[piodul]

@wmitros please take a look at it