Add bolt+routing with automatic failover and retry support

.github/workflows/ci.yml

@@ -104,7 +104,8 @@ jobs:
   build_and_test_linux:
     if: ${{ github.event_name == 'pull_request' || (github.event_name == 'workflow_dispatch' && inputs.linux) }}
     strategy:
-     matrix:
+      fail-fast: false
+      matrix:
         platform: [ubuntu-24.04, fedora-41]
         mgversion: ["latest"]
         packages: ["gcc g++ clang cmake git"]
@@ -128,7 +129,7 @@ jobs:
           docker network create ${{ env.MEMGRAPH_NETWORK }} || true
           platform="${{ matrix.platform }}"
           tag=${platform//-/:}
-          docker run -d --rm --name testcontainer --network ${{ env.MEMGRAPH_NETWORK }} "$tag" sleep infinity      
+          docker run -d --rm --name testcontainer --network ${{ env.MEMGRAPH_NETWORK }} "$tag" sleep infinity
 
       - name: Install environment
         run: |
@@ -159,13 +160,40 @@ jobs:
       - name: Load and run Memgraph Docker image
         run: |
           docker load -i memgraph-docker.tar.gz
-          docker run -d --rm --name memgraphcontainer --network ${{ env.MEMGRAPH_NETWORK }} -p 7687:7687 memgraph/memgraph --telemetry-enabled=false 
+          docker run -d --rm --name memgraphcontainer --network ${{ env.MEMGRAPH_NETWORK }} -p 7687:7687 memgraph/memgraph --telemetry-enabled=false
           rm memgraph-docker.tar.gz
           sleep 5
           docker logs memgraphcontainer
 
+      - name: Run Memgraph HA Cluster
+        env:
+          MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
+          MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
+        run: |
+          # High availability is a Memgraph Enterprise feature. The license
+          # secrets are unavailable on PRs from forks, so skip the cluster (and
+          # the routing tests, which GTEST_SKIP when the coordinator env vars are
+          # unset) with a warning rather than failing the whole build.
+          if [ -z "$MEMGRAPH_ENTERPRISE_LICENSE" ] || [ -z "$MEMGRAPH_ORGANIZATION_NAME" ]; then
+            echo "::warning::Memgraph enterprise license secrets are not set; skipping the HA cluster and the routing tests."
+            exit 0
+          fi
+
+          MEMGRAPH_HA_NETWORK="${{ env.MEMGRAPH_NETWORK }}" ./tool/ha_cluster.sh start
+
+          # Tell the test steps to run the routing tests against mg-coord1.
+          echo "MEMGRAPH_HA_COORDINATOR_HOST=127.0.0.1" >> "$GITHUB_ENV"
+          echo "MEMGRAPH_HA_COORDINATOR_PORT=7691" >> "$GITHUB_ENV"
+
       - name: Build with gcc, test and install mgclient
         run: |
+          # Forward the HA coordinator env vars into the container only when the
+          # cluster is up; when unset the routing tests GTEST_SKIP.
+          ha_env=()
+          if [ -n "${MEMGRAPH_HA_COORDINATOR_HOST:-}" ]; then
+            ha_env=(-e "MEMGRAPH_HA_COORDINATOR_HOST=${MEMGRAPH_HA_COORDINATOR_HOST}" \
+                    -e "MEMGRAPH_HA_COORDINATOR_PORT=${MEMGRAPH_HA_COORDINATOR_PORT}")
+          fi
           cmake_configure="cmake \
             -DCMAKE_C_COMPILER=gcc${{ matrix.gcc-postfix }} \
             -DCMAKE_CXX_COMPILER=g++${{ matrix.gcc-postfix }} \
@@ -176,16 +204,23 @@ jobs:
             -DCPP_WARNINGS_AS_ERRORS=ON \
             .."
 
-          docker exec -i testcontainer bash -c "
-            mkdir /mgclient/build-gcc && 
-            cd /mgclient/build-gcc && 
-            $cmake_configure && 
-            cmake --build . --parallel && 
-            ctest --output-on-failure && 
+          docker exec -i "${ha_env[@]}" testcontainer bash -c "
+            mkdir /mgclient/build-gcc &&
+            cd /mgclient/build-gcc &&
+            $cmake_configure &&
+            cmake --build . --parallel &&
+            ctest --output-on-failure &&
             make install"
-          
+
       - name: Build with clang, test and install mgclient
         run: |
+          # Forward the HA coordinator env vars into the container only when the
+          # cluster is up; when unset the routing tests GTEST_SKIP.
+          ha_env=()
+          if [ -n "${MEMGRAPH_HA_COORDINATOR_HOST:-}" ]; then
+            ha_env=(-e "MEMGRAPH_HA_COORDINATOR_HOST=${MEMGRAPH_HA_COORDINATOR_HOST}" \
+                    -e "MEMGRAPH_HA_COORDINATOR_PORT=${MEMGRAPH_HA_COORDINATOR_PORT}")
+          fi
           cmake_configure="cmake \
             -DCMAKE_C_COMPILER=clang${{ matrix.clang-postfix }} \
             -DCMAKE_CXX_COMPILER=clang++${{ matrix.clang-postfix }} \
@@ -196,19 +231,30 @@ jobs:
             -DCPP_WARNINGS_AS_ERRORS=ON \
           .."
 
-          docker exec -i testcontainer bash -c "
-            mkdir /mgclient/build-clang && 
-            cd /mgclient/build-clang && 
-            $cmake_configure && 
-            cmake --build . --parallel && 
-            ctest --output-on-failure && 
+          docker exec -i "${ha_env[@]}" testcontainer bash -c "
+            mkdir /mgclient/build-clang &&
+            cd /mgclient/build-clang &&
+            $cmake_configure &&
+            cmake --build . --parallel &&
+            ctest --output-on-failure &&
             make install"
 
+      - name: Run routing example
+        run: |
+          if [[ -z "${MEMGRAPH_HA_COORDINATOR_HOST}}" || -z "${MEMGRAPH_HA_COORDINATOR_PORT}" ]]; then
+            echo "::notice::Skipping HA routing example"
+            exit 0
+          fi
+          docker exec -i testcontainer \
+            /mgclient/build-gcc/tests/example_routing_c \
+            "$MEMGRAPH_HA_COORDINATOR_HOST" "$MEMGRAPH_HA_COORDINATOR_PORT"
+
       - name: Cleanup
         if: always()
         run: |
           docker stop testcontainer || echo "testcontainer already stopped"
           docker stop memgraphcontainer || echo "memgraphcontainer already stopped"
+          ./tool/ha_cluster.sh stop || echo "HA cluster already stopped"
           docker wait testcontainer || true
           docker wait memgraphcontainer || true
           docker rm "${{ env.MEMGRAPH_NETWORK }}" || echo "network already removed"

examples/routing.c

@@ -0,0 +1,193 @@
+// Copyright (c) 2016-2026 Memgraph Ltd. [https://memgraph.com]
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Client-side routing against a Memgraph high-availability (HA) cluster.
+//
+// You give the router the address of a *coordinator*. It fetches the cluster's
+// routing table (which instance is the main, which are replicas), caches it
+// until its TTL expires, and hands out connections to the right instance:
+// writes go to the main, reads to a replica.
+//
+// The managed helpers -- mg_router_execute_write / mg_router_execute_read --
+// go one step further: they run your unit of work and, if the cluster is
+// briefly in flux (for example while a new main is being elected during a
+// failover), they refresh the routing table and retry with exponential
+// back-off. So a write issued the instant the old main goes down still lands
+// on the new main once it is elected, without any special handling in your
+// code.
+//
+// Usage: routing <coordinator-host> <coordinator-port> [username] [password]
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <mgclient.h>
+
+// Drain any remaining rows in the current result stream. Returns 0 on success,
+// or a negative MG_ERROR_ code if a fetch failed.
+static int drain(mg_session *session) {
+  mg_result *result;
+  int status;
+  while ((status = mg_session_fetch(session, &result)) == 1) {
+  }
+  return status;
+}
+
+// A managed *write*: create/update a greeting node.
+//
+// The router owns the transaction around this callback, so the work must NOT
+// begin/commit it -- it only issues the query. Because the work may be called
+// more than once (on retry), it is written to be idempotent: MERGE rather than
+// CREATE, so a retry never duplicates the node.
+static int write_greeting(mg_session *session, void *data) {
+  (void)data;
+  int status = mg_session_run(
+      session, "MERGE (n:Greeting {id: 1}) SET n.text = 'hello from routing'",
+      NULL, NULL, NULL, NULL);
+  if (status != 0) {
+    return status;
+  }
+  if ((status = mg_session_pull(session, NULL)) != 0) {
+    return status;
+  }
+  return drain(session);
+}
+
+// A managed *read*: read the greeting text back into the caller's buffer.
+static int read_greeting(mg_session *session, void *data) {
+  char *out = (char *)data;
+  int status =
+      mg_session_run(session, "MATCH (n:Greeting {id: 1}) RETURN n.text", NULL,
+                     NULL, NULL, NULL);
+  if (status != 0) {
+    return status;
+  }
+  if ((status = mg_session_pull(session, NULL)) != 0) {
+    return status;
+  }
+
+  mg_result *result;
+  while ((status = mg_session_fetch(session, &result)) == 1) {
+    const mg_list *row = mg_result_row(result);
+    if (mg_list_size(row) > 0) {
+      const mg_value *value = mg_list_at(row, 0);
+      if (mg_value_get_type(value) == MG_VALUE_TYPE_STRING) {
+        const mg_string *text = mg_value_string(value);
+        snprintf(out, 256, "%.*s", (int)mg_string_size(text),
+                 mg_string_data(text));
+      }
+    }
+  }
+  return status;  // 0 once the stream is fully consumed
+}
+
+static void print_routing_table(mg_router *router) {
+  if (mg_router_refresh(router) != 0) {
+    printf("could not fetch routing table: %s\n", mg_router_error(router));
+    return;
+  }
+  const mg_routing_table *table = mg_router_routing_table(router);
+  printf("routing table (ttl %lld s):\n",
+         (long long)mg_routing_table_ttl(table));
+
+  const struct {
+    enum mg_routing_role role;
+    const char *name;
+  } roles[] = {{MG_ROUTING_ROLE_WRITE, "WRITE"},
+               {MG_ROUTING_ROLE_READ, "READ"},
+               {MG_ROUTING_ROLE_ROUTE, "ROUTE"}};
+  for (size_t r = 0; r < sizeof(roles) / sizeof(roles[0]); ++r) {
+    uint32_t count = mg_routing_table_address_count(table, roles[r].role);
+    for (uint32_t i = 0; i < count; ++i) {
+      printf("  %-5s %s\n", roles[r].name,
+             mg_routing_table_address_at(table, roles[r].role, i));
+    }
+  }
+}
+
+int main(int argc, char *argv[]) {
+  if (argc < 3 || argc > 5) {
+    fprintf(stderr,
+            "Usage: %s <coordinator-host> <coordinator-port> "
+            "[username] [password]\n",
+            argv[0]);
+    return 1;
+  }
+
+  mg_init();
+  printf("mgclient version: %s\n", mg_client_version());
+
+  // The seed coordinator, plus the connection template (auth, SSL, ...) that
+  // the router reuses for every routed connection it opens.
+  mg_session_params *params = mg_session_params_make();
+  mg_session_params_set_host(params, argv[1]);
+  mg_session_params_set_port(params, (uint16_t)atoi(argv[2]));
+  mg_session_params_set_sslmode(params, MG_SSLMODE_DISABLE);
+  if (argc >= 4) {
+    mg_session_params_set_username(params, argv[3]);
+  }
+  if (argc >= 5) {
+    mg_session_params_set_password(params, argv[4]);
+  }
+
+  mg_router_config *config = mg_router_config_make();
+  mg_router_config_set_session_params(config, params);
+  // The managed-transaction retry budget. These are the defaults, shown here
+  // only to make them explicit: up to 8 attempts, backing off 1s, 2s, 4s, ...
+  // capped at 15s between them.
+  mg_router_config_set_max_retries(config, 8);
+  mg_router_config_set_retry_backoff(config, 1.0, 15.0);
+  // If the cluster advertises addresses the client can't reach directly, set an
+  // address resolver here with mg_router_config_set_resolver(). Omitted for
+  // simplicity: advertised addresses are used as-is.
+
+  mg_router *router = mg_router_make(config);
+  mg_session_params_destroy(params);
+  mg_router_config_destroy(config);
+  if (!router) {
+    fprintf(stderr, "failed to create router\n");
+    mg_finalize();
+    return 1;
+  }
+
+  int exit_code = 0;
+
+  // Show the topology the coordinator reports.
+  print_routing_table(router);
+
+  // Managed write -> routed to the main, wrapped in a transaction, and retried
+  // (with a routing refresh) across a failover.
+  if (mg_router_execute_write(router, write_greeting, NULL) != 0) {
+    printf("write failed: %s\n", mg_router_error(router));
+    exit_code = 1;
+    goto cleanup;
+  }
+  printf("wrote greeting to the main\n");
+
+  // Managed read -> routed to a replica (round-robined across replicas on
+  // successive reads). Note: under ASYNC replication a replica may momentarily
+  // lag the main, so a freshly written value might not be visible yet.
+  char greeting[256] = "";
+  if (mg_router_execute_read(router, read_greeting, greeting) != 0) {
+    printf("read failed: %s\n", mg_router_error(router));
+    exit_code = 1;
+    goto cleanup;
+  }
+  printf("read greeting from a replica: \"%s\"\n", greeting);
+
+cleanup:
+  mg_router_destroy(router);
+  mg_finalize();
+  return exit_code;
+}