Replication

KVNode streams changes from a parent node to child nodes via an NDJSON-based replication protocol.

How Replication Works

  1. A client writes a key-value pair to a node.
  2. The receiving node persists the write in its WAL and applies it to the in-memory store.
  3. If the node has a parent configured, it forwards the write upstream via the /v1/replicate endpoint.
  4. If the node has children, it broadcasts the write to all connected replicas over the replication stream.
  5. Each replica applies the change to its own store, maintaining an eventually consistent copy.

Replication Protocol

The replication endpoint (POST /v1/replicate) uses NDJSON (Newline-Delimited JSON) as the wire format. The protocol supports these command types:

Command Description
SET A key was created or updated
DELETE A key was removed
BASELINE_BEGIN Start of a full collection snapshot
BASELINE_END End of a full collection snapshot
PING Keep-alive heartbeat

Example: Replication Stream

1{"type":"BASELINE_BEGIN","collection":"my-collection","snapshotSeq":1}
2{"type":"SET","collection":"my-collection","seq":1,"key":"user:alice","value":{"role":"admin"},"version":1,"timestamp":1700000000}
3{"type":"BASELINE_END","collection":"my-collection","snapshotSeq":1}
4{"type":"SET","collection":"my-collection","seq":2,"key":"user:bob","value":{"role":"user"},"version":1,"timestamp":1700000001}
5{"type":"PING","timestamp":1700000010}

Node Status

The /v1/status endpoint exposes the replication topology:

curl "https://api.hola.cloud/v1/status" \
  -H "apikey: your-api-key" \
  -H "secret: your-api-secret"
GET /v1/status HTTP/1.1
Host: api.hola.cloud
apikey: your-api-key
secret: your-api-secret
package main

import (
	"fmt"
	"io"
	"net/http"
)

func main() {
	req, err := http.NewRequest("GET", "https://api.hola.cloud/v1/status", nil)
	if err != nil {
		panic(err)
	}
	req.Header.Set("apikey", "your-api-key")
	req.Header.Set("secret", "your-api-secret")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	responseBody, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}
	fmt.Println(string(responseBody))
}
<?php
$ch = curl_init();

curl_setopt_array($ch, [
    CURLOPT_URL => 'https://api.hola.cloud/v1/status',
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_CUSTOMREQUEST => 'GET',
    CURLOPT_HTTPHEADER => [
        'apikey: your-api-key',
        'secret: your-api-secret',
    ],
]);

$response = curl_exec($ch);
if ($response === false) {
    throw new Exception(curl_error($ch));
}
curl_close($ch);

echo $response;
import requests

headers = {
    "apikey": "your-api-key",
    "secret": "your-api-secret",
}

response = requests.request(
    "GET",
    "https://api.hola.cloud/v1/status",
    headers=headers
)

print(response.text)
const response = await fetch("https://api.hola.cloud/v1/status", {
  method: "GET",
  headers: {
    "apikey": "your-api-key",
    "secret": "your-api-secret"
  }
});

console.log(await response.text());
const response = await fetch("https://api.hola.cloud/v1/status", {
  method: "GET",
  headers: {
    "apikey": "your-api-key",
    "secret": "your-api-secret"
  }
});

const text = await response.text();
console.log(text);
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

public class Main {
    public static void main(String[] args) throws Exception {
        var request = HttpRequest.newBuilder()
            .uri(URI.create("https://api.hola.cloud/v1/status"))
            .method("GET", HttpRequest.BodyPublishers.noBody())
            .header("apikey", "your-api-key")
            .header("secret", "your-api-secret")
            .build();

        var response = HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofString());
        System.out.println(response.body());
    }
}

Response:

 1{
 2  "node": "node-1",
 3  "role": "primary",
 4  "parent": "",
 5  "collections": {
 6    "my-collection": {
 7      "keys": 42,
 8      "lastSeq": 128,
 9      "walSizeBytes": 65536
10    }
11  },
12  "children": 2,
13  "uptimeSeconds": 86400,
14  "replication": {
15    "enabled": true,
16    "parent_connected": false
17  }
18}

Node Metrics

Monitor performance with the /v1/metrics endpoint:

curl "https://api.hola.cloud/v1/metrics" \
  -H "apikey: your-api-key" \
  -H "secret: your-api-secret"
GET /v1/metrics HTTP/1.1
Host: api.hola.cloud
apikey: your-api-key
secret: your-api-secret
package main

import (
	"fmt"
	"io"
	"net/http"
)

func main() {
	req, err := http.NewRequest("GET", "https://api.hola.cloud/v1/metrics", nil)
	if err != nil {
		panic(err)
	}
	req.Header.Set("apikey", "your-api-key")
	req.Header.Set("secret", "your-api-secret")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	responseBody, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}
	fmt.Println(string(responseBody))
}
<?php
$ch = curl_init();

curl_setopt_array($ch, [
    CURLOPT_URL => 'https://api.hola.cloud/v1/metrics',
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_CUSTOMREQUEST => 'GET',
    CURLOPT_HTTPHEADER => [
        'apikey: your-api-key',
        'secret: your-api-secret',
    ],
]);

$response = curl_exec($ch);
if ($response === false) {
    throw new Exception(curl_error($ch));
}
curl_close($ch);

echo $response;
import requests

headers = {
    "apikey": "your-api-key",
    "secret": "your-api-secret",
}

response = requests.request(
    "GET",
    "https://api.hola.cloud/v1/metrics",
    headers=headers
)

print(response.text)
const response = await fetch("https://api.hola.cloud/v1/metrics", {
  method: "GET",
  headers: {
    "apikey": "your-api-key",
    "secret": "your-api-secret"
  }
});

console.log(await response.text());
const response = await fetch("https://api.hola.cloud/v1/metrics", {
  method: "GET",
  headers: {
    "apikey": "your-api-key",
    "secret": "your-api-secret"
  }
});

const text = await response.text();
console.log(text);
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

public class Main {
    public static void main(String[] args) throws Exception {
        var request = HttpRequest.newBuilder()
            .uri(URI.create("https://api.hola.cloud/v1/metrics"))
            .method("GET", HttpRequest.BodyPublishers.noBody())
            .header("apikey", "your-api-key")
            .header("secret", "your-api-secret")
            .build();

        var response = HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofString());
        System.out.println(response.body());
    }
}

Response:

1{
2  "writes_total": 15000,
3  "reads_total": 82000,
4  "replication_commands_sent_total": 15000,
5  "replication_commands_received_total": 0,
6  "children_connected": 2,
7  "parent_connected": false
8}

Readiness Checks

The /readyz endpoint verifies that a node is ready to serve traffic. For replicas, this means the parent connection must be established and the WAL fully replayed. For primaries, readiness is immediate after startup.

curl "https://api.hola.cloud/readyz"
GET /readyz HTTP/1.1
Host: api.hola.cloud
package main

import (
	"fmt"
	"io"
	"net/http"
)

func main() {
	req, err := http.NewRequest("GET", "https://api.hola.cloud/readyz", nil)
	if err != nil {
		panic(err)
	}

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	responseBody, err := io.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}
	fmt.Println(string(responseBody))
}
<?php
$ch = curl_init();

curl_setopt_array($ch, [
    CURLOPT_URL => 'https://api.hola.cloud/readyz',
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_CUSTOMREQUEST => 'GET',
]);

$response = curl_exec($ch);
if ($response === false) {
    throw new Exception(curl_error($ch));
}
curl_close($ch);

echo $response;
import requests

response = requests.request(
    "GET",
    "https://api.hola.cloud/readyz"
)

print(response.text)
const response = await fetch("https://api.hola.cloud/readyz", {
  method: "GET"
});

console.log(await response.text());
const response = await fetch("https://api.hola.cloud/readyz", {
  method: "GET"
});

const text = await response.text();
console.log(text);
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

public class Main {
    public static void main(String[] args) throws Exception {
        var request = HttpRequest.newBuilder()
            .uri(URI.create("https://api.hola.cloud/readyz"))
            .method("GET", HttpRequest.BodyPublishers.noBody())
            .build();

        var response = HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofString());
        System.out.println(response.body());
    }
}

Response when not ready:

1{"ok":false,"node":"node-2","role":"replica","ready":false,"checks":{"wal_replayed":true,"parent_connected":false}}

Pluggable WAL Backends

KVNode supports multiple WAL backends, selectable at startup:

Backend Description Use Case
Memory In-process buffer (non-durable) Development, testing, ephemeral caches
Kafka Durable log via Apache Kafka High-throughput production deployments
PostgreSQL WAL stored in PostgreSQL tables Environments already using Postgres
Redis WAL stored in Redis streams Low-latency, Redis-centric architectures
MongoDB WAL stored in MongoDB collections MongoDB-centric deployments

Each backend implements the same WAL interface, so switching between them requires no application-level changes.

Comments

Leave a comment