← back to Exo
add forwarder service
596d9fc9d0147a86eaffe39f1fae0d2519661ded · 2025-07-22 20:53:26 +0100 · Gelu Vrabie
Co-authored-by: Gelu Vrabie <gelu@exolabs.net>
Files touched
M .gitignoreM flake.lockM flake.nixA networking/forwarder/benchmark.shA networking/forwarder/go.modA networking/forwarder/go.sumA networking/forwarder/main.goA networking/forwarder/src/config.goA networking/forwarder/src/forwarder.goA networking/forwarder/src/forwarder_test.goA networking/forwarder/src/identity.goA networking/forwarder/src/libp2p.goA networking/forwarder/src/libp2p_test.goA networking/forwarder/src/schema.goA networking/forwarder/src/sqlite.goA networking/forwarder/src/sqlite_test.goA networking/forwarder/src/state_store.goA networking/forwarder/src/state_store_test.goR100 networking/.gitignore networking/topology/.gitignoreR100 networking/Cargo.lock networking/topology/Cargo.lockR100 networking/Cargo.toml networking/topology/Cargo.tomlR095 networking/pyproject.toml networking/topology/pyproject.tomlR100 networking/src/lib.rs networking/topology/src/lib.rsR100 networking/src/networking/__init__.py networking/topology/src/networking/__init__.pyR100 networking/src/networking/_core.pyi networking/topology/src/networking/_core.pyiM pyproject.tomlM uv.lockM worker/pyproject.toml
Diff
commit 596d9fc9d0147a86eaffe39f1fae0d2519661ded
Author: Gelu Vrabie <gelu.vrabie.univ@gmail.com>
Date: Tue Jul 22 20:53:26 2025 +0100
add forwarder service
Co-authored-by: Gelu Vrabie <gelu@exolabs.net>
---
.gitignore | 2 +
flake.lock | 34 ++
flake.nix | 96 +--
networking/forwarder/benchmark.sh | 106 ++++
networking/forwarder/go.mod | 127 ++++
networking/forwarder/go.sum | 555 ++++++++++++++++++
networking/forwarder/main.go | 59 ++
networking/forwarder/src/config.go | 91 +++
networking/forwarder/src/forwarder.go | 133 +++++
networking/forwarder/src/forwarder_test.go | 474 +++++++++++++++
networking/forwarder/src/identity.go | 29 +
networking/forwarder/src/libp2p.go | 414 +++++++++++++
networking/forwarder/src/libp2p_test.go | 175 ++++++
networking/forwarder/src/schema.go | 72 +++
networking/forwarder/src/sqlite.go | 649 +++++++++++++++++++++
networking/forwarder/src/sqlite_test.go | 236 ++++++++
networking/forwarder/src/state_store.go | 240 ++++++++
networking/forwarder/src/state_store_test.go | 283 +++++++++
networking/{ => topology}/.gitignore | 0
networking/{ => topology}/Cargo.lock | 0
networking/{ => topology}/Cargo.toml | 0
networking/{ => topology}/pyproject.toml | 1 -
networking/{ => topology}/src/lib.rs | 0
.../{ => topology}/src/networking/__init__.py | 0
networking/{ => topology}/src/networking/_core.pyi | 0
pyproject.toml | 2 +-
uv.lock | 48 +-
worker/pyproject.toml | 2 +-
28 files changed, 3762 insertions(+), 66 deletions(-)
diff --git a/.gitignore b/.gitignore
index 8275d34c..e9a1c1ff 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,4 +1,6 @@
*/__pycache__
__pycache__
+networking/target/*
+*.so
hosts_*.json
\ No newline at end of file
diff --git a/flake.lock b/flake.lock
index 933a2f61..b2380393 100644
--- a/flake.lock
+++ b/flake.lock
@@ -1,5 +1,23 @@
{
"nodes": {
+ "flake-utils": {
+ "inputs": {
+ "systems": "systems"
+ },
+ "locked": {
+ "lastModified": 1731533236,
+ "narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
+ "owner": "numtide",
+ "repo": "flake-utils",
+ "rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
+ "type": "github"
+ },
+ "original": {
+ "owner": "numtide",
+ "repo": "flake-utils",
+ "type": "github"
+ }
+ },
"nixpkgs": {
"locked": {
"lastModified": 1749794982,
@@ -18,8 +36,24 @@
},
"root": {
"inputs": {
+ "flake-utils": "flake-utils",
"nixpkgs": "nixpkgs"
}
+ },
+ "systems": {
+ "locked": {
+ "lastModified": 1681028828,
+ "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+ "owner": "nix-systems",
+ "repo": "default",
+ "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+ "type": "github"
+ },
+ "original": {
+ "owner": "nix-systems",
+ "repo": "default",
+ "type": "github"
+ }
}
},
"root": "root",
diff --git a/flake.nix b/flake.nix
index 006af63c..44f676ac 100644
--- a/flake.nix
+++ b/flake.nix
@@ -1,46 +1,64 @@
{
- description = "The development environment for Exo";
+ description = "Exo development flake";
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+ flake-utils.url = "github:numtide/flake-utils";
};
- outputs = { self, nixpkgs }:
- let
- supportedSystems = [ "x86_64-linux" "aarch64-linux" "x86_64-darwin" "aarch64-darwin" ];
- forAllSystems = nixpkgs.lib.genAttrs supportedSystems;
- in
- {
- devShells = forAllSystems (system:
- let
- pkgs = import nixpkgs { inherit system; };
- in
- {
- default = pkgs.mkShell {
- packages = [
- pkgs.python313
- pkgs.uv
- pkgs.just
- pkgs.protobuf
- pkgs.rustc
- pkgs.cargo
- pkgs.basedpyright
- pkgs.ruff
- ];
- };
- }
- );
-
- apps = forAllSystems (system:
- let
- pkgs = import nixpkgs { inherit system; };
- in
- {
- python-lsp = {
- type = "app";
- program = "${pkgs.basedpyright}/bin/basedpyright-langserver";
- };
- }
- );
- };
+ outputs = { self, nixpkgs, flake-utils }:
+ flake-utils.lib.eachDefaultSystem (system:
+ let
+ pkgs = import nixpkgs { inherit system; };
+
+ # Go 1.23 compiler – align with go.mod
+ go = pkgs.go_1_23;
+
+ # Build the networking/forwarder Go utility.
+ forwarder = pkgs.buildGoModule {
+ pname = "exo-forwarder";
+ version = "0.1.0";
+ src = ./networking/forwarder;
+
+ vendorHash = "sha256-BXIGg2QYqHDz2TNe8hLAGC6jVlffp9766H+WdkkuVgA=";
+
+ # Only the main package at the repository root needs building.
+ subPackages = [ "." ];
+ };
+ in
+ {
+ packages = {
+ inherit forwarder;
+ default = forwarder;
+ };
+
+ apps.forwarder = {
+ type = "app";
+ program = "${forwarder}/bin/forwarder";
+ };
+ apps.python-lsp = {
+ type = "app";
+ program = "${pkgs.basedpyright}/bin/basedpyright-langserver";
+ };
+ apps.default = self.apps.${system}.forwarder;
+
+ devShells.default = pkgs.mkShell {
+ packages = [
+ pkgs.python313
+ pkgs.uv
+ pkgs.just
+ pkgs.protobuf
+ pkgs.rustc
+ pkgs.cargo
+ pkgs.basedpyright
+ pkgs.ruff
+ go
+ ];
+
+ shellHook = ''
+ export GOPATH=$(mktemp -d)
+ '';
+ };
+ }
+ );
}
\ No newline at end of file
diff --git a/networking/forwarder/benchmark.sh b/networking/forwarder/benchmark.sh
new file mode 100755
index 00000000..72f4682b
--- /dev/null
+++ b/networking/forwarder/benchmark.sh
@@ -0,0 +1,106 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+NUM_RECORDS="${1:-10000}"
+BATCH_SIZE="${2:-100}"
+
+echo "Running burst benchmark with $NUM_RECORDS records in batches of $BATCH_SIZE..."
+
+# Build the forwarder binary
+BIN_PATH="$(pwd)/forwarder_bin"
+BUILD_TMPDIR="$(mktemp -d 2>/dev/null || mktemp -d -t forwarder-build)"
+export TMPDIR="$BUILD_TMPDIR"
+
+pushd . >/dev/null
+go build -o "$BIN_PATH" .
+popd >/dev/null
+
+# Temporary workspace
+TMP_DIR="$(mktemp -d 2>/dev/null || mktemp -d -t forwarder-burst)"
+SRC_DB="$TMP_DIR/src.db"
+DST_DB="$TMP_DIR/dst.db"
+TABLE="records"
+TOPIC="burst_topic_$$"
+
+# Cleanup function
+cleanup() {
+ echo "Cleaning up…"
+ kill "${PID1:-}" "${PID2:-}" 2>/dev/null || true
+ wait "${PID1:-}" "${PID2:-}" 2>/dev/null || true
+ rm -rf "$TMP_DIR" "$BIN_PATH" "$BUILD_TMPDIR"
+}
+trap cleanup EXIT
+
+# Create databases with WAL mode
+sqlite3 "$SRC_DB" <<SQL
+.timeout 5000
+PRAGMA journal_mode=WAL;
+SQL
+
+sqlite3 "$DST_DB" <<SQL
+.timeout 5000
+PRAGMA journal_mode=WAL;
+SQL
+
+# Start forwarder nodes
+"$BIN_PATH" -node-id node1 "sqlite:${SRC_DB}:${TABLE}|libp2p:${TOPIC}" >"$TMP_DIR/node1.log" 2>&1 &
+PID1=$!
+
+"$BIN_PATH" -node-id node2 "libp2p:${TOPIC}|sqlite:${DST_DB}:${TABLE}" >"$TMP_DIR/node2.log" 2>&1 &
+PID2=$!
+
+# Give nodes time to start
+sleep 3
+
+echo "Inserting $NUM_RECORDS records in batches of $BATCH_SIZE..."
+START_NS=$(date +%s%N)
+
+# Insert records in batches for high throughput
+for batch_start in $(seq 1 $BATCH_SIZE $NUM_RECORDS); do
+ batch_end=$((batch_start + BATCH_SIZE - 1))
+ if [ $batch_end -gt $NUM_RECORDS ]; then
+ batch_end=$NUM_RECORDS
+ fi
+
+ # Build values for batch insert
+ values=""
+ for i in $(seq $batch_start $batch_end); do
+ if [ -n "$values" ]; then
+ values="$values,"
+ fi
+ values="$values('seednode','seedpath',$i,datetime('now'),'{}')"
+ done
+
+ # Insert batch
+ sqlite3 -cmd ".timeout 5000" "$SRC_DB" \
+ "INSERT INTO ${TABLE} (source_node_id, source_path, source_row_id, source_timestamp, data) VALUES $values;"
+
+ # Small delay to prevent overwhelming
+ sleep 0.01
+done
+
+echo "Waiting for destination to catch up..."
+
+# Wait for completion
+while true; do
+ dest_count=$(sqlite3 -cmd ".timeout 5000" "$DST_DB" "SELECT IFNULL(COUNT(*),0) FROM ${TABLE};" 2>/dev/null || echo 0)
+ if [[ "$dest_count" -ge "$NUM_RECORDS" ]]; then
+ break
+ fi
+ echo "Progress: $dest_count / $NUM_RECORDS"
+ sleep 1
+done
+
+END_NS=$(date +%s%N)
+DURATION_NS=$((END_NS-START_NS))
+THROUGHPUT=$(echo "scale=2; $NUM_RECORDS*1000000000/$DURATION_NS" | bc)
+
+echo "Forwarded $NUM_RECORDS records in $(printf '%.2f' "$(echo "$DURATION_NS/1000000000" | bc -l)") seconds — $THROUGHPUT records/s"
+
+# Show some logs
+echo ""
+echo "=== Node1 Log (last 10 lines) ==="
+tail -10 "$TMP_DIR/node1.log"
+echo ""
+echo "=== Node2 Log (last 10 lines) ==="
+tail -10 "$TMP_DIR/node2.log"
\ No newline at end of file
diff --git a/networking/forwarder/go.mod b/networking/forwarder/go.mod
new file mode 100644
index 00000000..b7100a6a
--- /dev/null
+++ b/networking/forwarder/go.mod
@@ -0,0 +1,127 @@
+module forwarder
+
+go 1.23
+
+toolchain go1.24.3
+
+replace forwarder/src => ./src
+
+require (
+ github.com/google/uuid v1.6.0
+ github.com/libp2p/go-libp2p v0.39.1
+ github.com/libp2p/go-libp2p-pubsub v0.14.2
+ github.com/mattn/go-sqlite3 v1.14.28
+ github.com/stretchr/testify v1.10.0
+)
+
+require (
+ github.com/benbjohnson/clock v1.3.5 // indirect
+ github.com/beorn7/perks v1.0.1 // indirect
+ github.com/cespare/xxhash/v2 v2.3.0 // indirect
+ github.com/containerd/cgroups v1.1.0 // indirect
+ github.com/coreos/go-systemd/v22 v22.5.0 // indirect
+ github.com/davecgh/go-spew v1.1.1 // indirect
+ github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect
+ github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 // indirect
+ github.com/docker/go-units v0.5.0 // indirect
+ github.com/elastic/gosigar v0.14.3 // indirect
+ github.com/flynn/noise v1.1.0 // indirect
+ github.com/francoispqt/gojay v1.2.13 // indirect
+ github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
+ github.com/godbus/dbus/v5 v5.1.0 // indirect
+ github.com/gogo/protobuf v1.3.2 // indirect
+ github.com/google/gopacket v1.1.19 // indirect
+ github.com/google/pprof v0.0.0-20250202011525-fc3143867406 // indirect
+ github.com/gorilla/websocket v1.5.3 // indirect
+ github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
+ github.com/huin/goupnp v1.3.0 // indirect
+ github.com/ipfs/go-cid v0.5.0 // indirect
+ github.com/ipfs/go-log/v2 v2.5.1 // indirect
+ github.com/jackpal/go-nat-pmp v1.0.2 // indirect
+ github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+ github.com/klauspost/compress v1.17.11 // indirect
+ github.com/klauspost/cpuid/v2 v2.2.9 // indirect
+ github.com/koron/go-ssdp v0.0.5 // indirect
+ github.com/libp2p/go-buffer-pool v0.1.0 // indirect
+ github.com/libp2p/go-flow-metrics v0.2.0 // indirect
+ github.com/libp2p/go-libp2p-asn-util v0.4.1 // indirect
+ github.com/libp2p/go-msgio v0.3.0 // indirect
+ github.com/libp2p/go-nat v0.2.0 // indirect
+ github.com/libp2p/go-netroute v0.2.2 // indirect
+ github.com/libp2p/go-reuseport v0.4.0 // indirect
+ github.com/libp2p/go-yamux/v4 v4.0.2 // indirect
+ github.com/libp2p/zeroconf/v2 v2.2.0 // indirect
+ github.com/marten-seemann/tcp v0.0.0-20210406111302-dfbc87cc63fd // indirect
+ github.com/mattn/go-isatty v0.0.20 // indirect
+ github.com/miekg/dns v1.1.63 // indirect
+ github.com/mikioh/tcpinfo v0.0.0-20190314235526-30a79bb1804b // indirect
+ github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc // indirect
+ github.com/minio/sha256-simd v1.0.1 // indirect
+ github.com/mr-tron/base58 v1.2.0 // indirect
+ github.com/multiformats/go-base32 v0.1.0 // indirect
+ github.com/multiformats/go-base36 v0.2.0 // indirect
+ github.com/multiformats/go-multiaddr v0.14.0 // indirect
+ github.com/multiformats/go-multiaddr-dns v0.4.1 // indirect
+ github.com/multiformats/go-multiaddr-fmt v0.1.0 // indirect
+ github.com/multiformats/go-multibase v0.2.0 // indirect
+ github.com/multiformats/go-multicodec v0.9.0 // indirect
+ github.com/multiformats/go-multihash v0.2.3 // indirect
+ github.com/multiformats/go-multistream v0.6.0 // indirect
+ github.com/multiformats/go-varint v0.0.7 // indirect
+ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+ github.com/onsi/ginkgo/v2 v2.22.2 // indirect
+ github.com/opencontainers/runtime-spec v1.2.0 // indirect
+ github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
+ github.com/pion/datachannel v1.5.10 // indirect
+ github.com/pion/dtls/v2 v2.2.12 // indirect
+ github.com/pion/dtls/v3 v3.0.4 // indirect
+ github.com/pion/ice/v2 v2.3.37 // indirect
+ github.com/pion/ice/v4 v4.0.6 // indirect
+ github.com/pion/interceptor v0.1.37 // indirect
+ github.com/pion/logging v0.2.3 // indirect
+ github.com/pion/mdns v0.0.12 // indirect
+ github.com/pion/mdns/v2 v2.0.7 // indirect
+ github.com/pion/randutil v0.1.0 // indirect
+ github.com/pion/rtcp v1.2.15 // indirect
+ github.com/pion/rtp v1.8.11 // indirect
+ github.com/pion/sctp v1.8.35 // indirect
+ github.com/pion/sdp/v3 v3.0.10 // indirect
+ github.com/pion/srtp/v3 v3.0.4 // indirect
+ github.com/pion/stun v0.6.1 // indirect
+ github.com/pion/stun/v3 v3.0.0 // indirect
+ github.com/pion/transport/v2 v2.2.10 // indirect
+ github.com/pion/transport/v3 v3.0.7 // indirect
+ github.com/pion/turn/v2 v2.1.6 // indirect
+ github.com/pion/turn/v4 v4.0.0 // indirect
+ github.com/pion/webrtc/v4 v4.0.8 // indirect
+ github.com/pkg/errors v0.9.1 // indirect
+ github.com/pmezard/go-difflib v1.0.0 // indirect
+ github.com/prometheus/client_golang v1.20.5 // indirect
+ github.com/prometheus/client_model v0.6.1 // indirect
+ github.com/prometheus/common v0.62.0 // indirect
+ github.com/prometheus/procfs v0.15.1 // indirect
+ github.com/quic-go/qpack v0.5.1 // indirect
+ github.com/quic-go/quic-go v0.49.0 // indirect
+ github.com/quic-go/webtransport-go v0.8.1-0.20241018022711-4ac2c9250e66 // indirect
+ github.com/raulk/go-watchdog v1.3.0 // indirect
+ github.com/spaolacci/murmur3 v1.1.0 // indirect
+ github.com/wlynxg/anet v0.0.5 // indirect
+ go.uber.org/dig v1.18.0 // indirect
+ go.uber.org/fx v1.23.0 // indirect
+ go.uber.org/mock v0.5.0 // indirect
+ go.uber.org/multierr v1.11.0 // indirect
+ go.uber.org/zap v1.27.0 // indirect
+ golang.org/x/crypto v0.32.0 // indirect
+ golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c // indirect
+ golang.org/x/mod v0.23.0 // indirect
+ golang.org/x/net v0.34.0 // indirect
+ golang.org/x/sync v0.11.0 // indirect
+ golang.org/x/sys v0.30.0 // indirect
+ golang.org/x/text v0.22.0 // indirect
+ golang.org/x/tools v0.29.0 // indirect
+ google.golang.org/protobuf v1.36.4 // indirect
+ gopkg.in/yaml.v3 v3.0.1 // indirect
+ lukechampine.com/blake3 v1.3.0 // indirect
+)
+
+// Remember to run `go mod tidy` after adding dependencies.
diff --git a/networking/forwarder/go.sum b/networking/forwarder/go.sum
new file mode 100644
index 00000000..75e179a9
--- /dev/null
+++ b/networking/forwarder/go.sum
@@ -0,0 +1,555 @@
+cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+cloud.google.com/go v0.31.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+cloud.google.com/go v0.37.0/go.mod h1:TS1dMSSfndXH133OKGwekG838Om/cQT0BUHV3HcBgoo=
+dmitri.shuralyov.com/app/changes v0.0.0-20180602232624-0a106ad413e3/go.mod h1:Yl+fi1br7+Rr3LqpNJf1/uxUdtRUV+Tnj0o93V2B9MU=
+dmitri.shuralyov.com/html/belt v0.0.0-20180602232347-f7d459c86be0/go.mod h1:JLBrvjyP0v+ecvNYvCpyZgu5/xkfAUhi6wJj28eUfSU=
+dmitri.shuralyov.com/service/change v0.0.0-20181023043359-a85b471d5412/go.mod h1:a1inKt/atXimZ4Mv927x+r7UpyzRUf4emIoiiSC2TN4=
+dmitri.shuralyov.com/state v0.0.0-20180228185332-28bcc343414c/go.mod h1:0PRwlb0D6DFvNNtx+9ybjezNCa8XF0xaYcETyp6rHWU=
+git.apache.org/thrift.git v0.0.0-20180902110319-2566ecd5d999/go.mod h1:fPE2ZNJGynbRyZ4dJvy6G277gSllfV2HJqblrnkyeyg=
+github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
+github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
+github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
+github.com/benbjohnson/clock v1.3.5 h1:VvXlSJBzZpA/zum6Sj74hxwYI2DIxRWuNIoXAzHZz5o=
+github.com/benbjohnson/clock v1.3.5/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
+github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/bradfitz/go-smtpd v0.0.0-20170404230938-deb6d6237625/go.mod h1:HYsPBTaaSFSlLx/70C2HPIMNZpVV8+vt/A+FMnYP11g=
+github.com/buger/jsonparser v0.0.0-20181115193947-bf1c66bbce23/go.mod h1:bbYlZJ7hK1yFx9hf58LP0zeX7UjIGs20ufpu3evjr+s=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cilium/ebpf v0.2.0/go.mod h1:To2CFviqOWL/M0gIMsvSMlqe7em/l1ALkX1PyjrX2Qs=
+github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
+github.com/containerd/cgroups v0.0.0-20201119153540-4cbc285b3327/go.mod h1:ZJeTFisyysqgcCdecO57Dj79RfL0LNeGiFUqLYQRYLE=
+github.com/containerd/cgroups v1.1.0 h1:v8rEWFl6EoqHB+swVNjVoCJE8o3jX7e8nqBGPLaDFBM=
+github.com/containerd/cgroups v1.1.0/go.mod h1:6ppBcbh/NOOUU+dMKrykgaBnK9lCIBxHqJDGwsa1mIw=
+github.com/coreos/go-systemd v0.0.0-20181012123002-c6f51f82210d/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
+github.com/coreos/go-systemd/v22 v22.1.0/go.mod h1:xO0FLkIi5MaZafQlIrOotqXZ90ih+1atmu1JpKERPPk=
+github.com/coreos/go-systemd/v22 v22.5.0 h1:RrqgGjYQKalulkV8NGVIfkXQf6YYmOyiJKk8iXXhfZs=
+github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
+github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
+github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c h1:pFUpOrbxDR6AkioZ1ySsx5yxlDQZ8stG2b88gTPxgJU=
+github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c/go.mod h1:6UhI8N9EjYm1c2odKpFpAYeR8dsBeM7PtzQhRgxRr9U=
+github.com/decred/dcrd/crypto/blake256 v1.0.1 h1:7PltbUIQB7u/FfZ39+DGa/ShuMyJ5ilcvdfma9wOH6Y=
+github.com/decred/dcrd/crypto/blake256 v1.0.1/go.mod h1:2OfgNZ5wDpcsFmHmCK5gZTPcCXqlm2ArzUIkw9czNJo=
+github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 h1:rpfIENRNNilwHwZeG5+P150SMrnNEcHYvcCuK6dPZSg=
+github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0/go.mod h1:v57UDF4pDQJcEfFUCRop3lJL149eHGSe9Jvczhzjo/0=
+github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
+github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
+github.com/elastic/gosigar v0.12.0/go.mod h1:iXRIGg2tLnu7LBdpqzyQfGDEidKCfWcCMS0WKyPWoMs=
+github.com/elastic/gosigar v0.14.3 h1:xwkKwPia+hSfg9GqrCUKYdId102m9qTJIIr7egmK/uo=
+github.com/elastic/gosigar v0.14.3/go.mod h1:iXRIGg2tLnu7LBdpqzyQfGDEidKCfWcCMS0WKyPWoMs=
+github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568/go.mod h1:xEzjJPgXI435gkrCt3MPfRiAkVrwSbHsst4LCFVfpJc=
+github.com/flynn/noise v1.1.0 h1:KjPQoQCEFdZDiP03phOvGi11+SVVhBG2wOWAorLsstg=
+github.com/flynn/noise v1.1.0/go.mod h1:xbMo+0i6+IGbYdJhF31t2eR1BIU0CYc12+BNAKwUTag=
+github.com/francoispqt/gojay v1.2.13 h1:d2m3sFjloqoIUQU3TsHBgj6qg/BVGlTBeHDUmyJnXKk=
+github.com/francoispqt/gojay v1.2.13/go.mod h1:ehT5mTG4ua4581f1++1WLG0vPdaA9HaiDsoyrBGkyDY=
+github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
+github.com/gliderlabs/ssh v0.1.1/go.mod h1:U7qILu1NlMHj9FlMhZLlkCdDnU1DBEAqr0aevW3Awn0=
+github.com/go-errors/errors v1.0.1/go.mod h1:f4zRHt4oKfwPJE5k8C9vpYG+aDHdBFUsgrm6/TyX73Q=
+github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
+github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
+github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
+github.com/godbus/dbus/v5 v5.0.3/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/godbus/dbus/v5 v5.1.0 h1:4KLkAxT3aOY8Li4FRJe/KvhoNFFxo0m6fNuFUO8QJUk=
+github.com/godbus/dbus/v5 v5.1.0/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
+github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
+github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
+github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
+github.com/golang/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:tluoj9z5200jBnyusfRPU2LqT6J+DAorxEvtC7LHB+E=
+github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
+github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
+github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
+github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-github v17.0.0+incompatible/go.mod h1:zLgOLi98H3fifZn+44m+umXrS52loVEgC2AApnigrVQ=
+github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
+github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
+github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
+github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
+github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
+github.com/google/pprof v0.0.0-20250202011525-fc3143867406 h1:wlQI2cYY0BsWmmPPAnxfQ8SDW0S3Jasn+4B8kXFxprg=
+github.com/google/pprof v0.0.0-20250202011525-fc3143867406/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144=
+github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/googleapis/gax-go v2.0.0+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
+github.com/googleapis/gax-go/v2 v2.0.3/go.mod h1:LLvjysVCY1JZeum8Z6l8qUty8fiNwE08qbEPm1M08qg=
+github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
+github.com/grpc-ecosystem/grpc-gateway v1.5.0/go.mod h1:RSKVYQBd5MCa4OVpNdGskqpgL2+G+NZTnrVHpWWfpdw=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
+github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
+github.com/huin/goupnp v1.3.0 h1:UvLUlWDNpoUdYzb2TCn+MuTWtcjXKSza2n6CBdQ0xXc=
+github.com/huin/goupnp v1.3.0/go.mod h1:gnGPsThkYa7bFi/KWmEysQRf48l2dvR5bxr2OFckNX8=
+github.com/ipfs/go-cid v0.5.0 h1:goEKKhaGm0ul11IHA7I6p1GmKz8kEYniqFopaB5Otwg=
+github.com/ipfs/go-cid v0.5.0/go.mod h1:0L7vmeNXpQpUS9vt+yEARkJ8rOg43DF3iPgn4GIN0mk=
+github.com/ipfs/go-log/v2 v2.5.1 h1:1XdUzF7048prq4aBjDQQ4SL5RxftpRGdXhNRwKSAlcY=
+github.com/ipfs/go-log/v2 v2.5.1/go.mod h1:prSpmC1Gpllc9UYWxDiZDreBYw7zp4Iqp1kOLU9U5UI=
+github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7BdWus=
+github.com/jackpal/go-nat-pmp v1.0.2/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=
+github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABoLk/+KKHggpk=
+github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
+github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
+github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
+github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
+github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
+github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
+github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
+github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
+github.com/klauspost/cpuid/v2 v2.2.9 h1:66ze0taIn2H33fBvCkXuv9BmCwDfafmiIVpKV9kKGuY=
+github.com/klauspost/cpuid/v2 v2.2.9/go.mod h1:rqkxqrZ1EhYM9G+hXH7YdowN5R5RGN6NK4QwQ3WMXF8=
+github.com/koron/go-ssdp v0.0.5 h1:E1iSMxIs4WqxTbIBLtmNBeOOC+1sCIXQeqTWVnpmwhk=
+github.com/koron/go-ssdp v0.0.5/go.mod h1:Qm59B7hpKpDqfyRNWRNr00jGwLdXjDyZh6y7rH6VS0w=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/pty v1.1.3/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/libp2p/go-buffer-pool v0.1.0 h1:oK4mSFcQz7cTQIfqbe4MIj9gLW+mnanjyFtc6cdF0Y8=
+github.com/libp2p/go-buffer-pool v0.1.0/go.mod h1:N+vh8gMqimBzdKkSMVuydVDq+UV5QTWy5HSiZacSbPg=
+github.com/libp2p/go-flow-metrics v0.2.0 h1:EIZzjmeOE6c8Dav0sNv35vhZxATIXWZg6j/C08XmmDw=
+github.com/libp2p/go-flow-metrics v0.2.0/go.mod h1:st3qqfu8+pMfh+9Mzqb2GTiwrAGjIPszEjZmtksN8Jc=
+github.com/libp2p/go-libp2p v0.39.1 h1:1Ur6rPCf3GR+g8jkrnaQaM0ha2IGespsnNlCqJLLALE=
+github.com/libp2p/go-libp2p v0.39.1/go.mod h1:3zicI8Lp7Isun+Afo/JOACUbbJqqR2owK6RQWFsVAbI=
+github.com/libp2p/go-libp2p-asn-util v0.4.1 h1:xqL7++IKD9TBFMgnLPZR6/6iYhawHKHl950SO9L6n94=
+github.com/libp2p/go-libp2p-asn-util v0.4.1/go.mod h1:d/NI6XZ9qxw67b4e+NgpQexCIiFYJjErASrYW4PFDN8=
+github.com/libp2p/go-libp2p-pubsub v0.14.2 h1:nT5lFHPQOFJcp9CW8hpKtvbpQNdl2udJuzLQWbgRum8=
+github.com/libp2p/go-libp2p-pubsub v0.14.2/go.mod h1:MKPU5vMI8RRFyTP0HfdsF9cLmL1nHAeJm44AxJGJx44=
+github.com/libp2p/go-libp2p-testing v0.12.0 h1:EPvBb4kKMWO29qP4mZGyhVzUyR25dvfUIK5WDu6iPUA=
+github.com/libp2p/go-libp2p-testing v0.12.0/go.mod h1:KcGDRXyN7sQCllucn1cOOS+Dmm7ujhfEyXQL5lvkcPg=
+github.com/libp2p/go-msgio v0.3.0 h1:mf3Z8B1xcFN314sWX+2vOTShIE0Mmn2TXn3YCUQGNj0=
+github.com/libp2p/go-msgio v0.3.0/go.mod h1:nyRM819GmVaF9LX3l03RMh10QdOroF++NBbxAb0mmDM=
+github.com/libp2p/go-nat v0.2.0 h1:Tyz+bUFAYqGyJ/ppPPymMGbIgNRH+WqC5QrT5fKrrGk=
+github.com/libp2p/go-nat v0.2.0/go.mod h1:3MJr+GRpRkyT65EpVPBstXLvOlAPzUVlG6Pwg9ohLJk=
+github.com/libp2p/go-netroute v0.2.2 h1:Dejd8cQ47Qx2kRABg6lPwknU7+nBnFRpko45/fFPuZ8=
+github.com/libp2p/go-netroute v0.2.2/go.mod h1:Rntq6jUAH0l9Gg17w5bFGhcC9a+vk4KNXs6s7IljKYE=
+github.com/libp2p/go-reuseport v0.4.0 h1:nR5KU7hD0WxXCJbmw7r2rhRYruNRl2koHw8fQscQm2s=
+github.com/libp2p/go-reuseport v0.4.0/go.mod h1:ZtI03j/wO5hZVDFo2jKywN6bYKWLOy8Se6DrI2E1cLU=
+github.com/libp2p/go-yamux/v4 v4.0.2 h1:nrLh89LN/LEiqcFiqdKDRHjGstN300C1269K/EX0CPU=
+github.com/libp2p/go-yamux/v4 v4.0.2/go.mod h1:C808cCRgOs1iBwY4S71T5oxgMxgLmqUw56qh4AeBW2o=
+github.com/libp2p/zeroconf/v2 v2.2.0 h1:Cup06Jv6u81HLhIj1KasuNM/RHHrJ8T7wOTS4+Tv53Q=
+github.com/libp2p/zeroconf/v2 v2.2.0/go.mod h1:fuJqLnUwZTshS3U/bMRJ3+ow/v9oid1n0DmyYyNO1Xs=
+github.com/lunixbochs/vtclean v1.0.0/go.mod h1:pHhQNgMf3btfWnGBVipUOjRYhoOsdGqdm/+2c2E2WMI=
+github.com/mailru/easyjson v0.0.0-20190312143242-1de009706dbe/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
+github.com/marten-seemann/tcp v0.0.0-20210406111302-dfbc87cc63fd h1:br0buuQ854V8u83wA0rVZ8ttrq5CpaPZdvrK0LP2lOk=
+github.com/marten-seemann/tcp v0.0.0-20210406111302-dfbc87cc63fd/go.mod h1:QuCEs1Nt24+FYQEqAAncTDPJIuGs+LxK1MCiFL25pMU=
+github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
+github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
+github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/mattn/go-sqlite3 v1.14.28 h1:ThEiQrnbtumT+QMknw63Befp/ce/nUPgBPMlRFEum7A=
+github.com/mattn/go-sqlite3 v1.14.28/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
+github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
+github.com/microcosm-cc/bluemonday v1.0.1/go.mod h1:hsXNsILzKxV+sX77C5b8FSuKF00vh2OMYv+xgHpAMF4=
+github.com/miekg/dns v1.1.43/go.mod h1:+evo5L0630/F6ca/Z9+GAqzhjGyn8/c+TBaOyfEl0V4=
+github.com/miekg/dns v1.1.63 h1:8M5aAw6OMZfFXTT7K5V0Eu5YiiL8l7nUAkyN6C9YwaY=
+github.com/miekg/dns v1.1.63/go.mod h1:6NGHfjhpmr5lt3XPLuyfDJi5AXbNIPM9PY6H6sF1Nfs=
+github.com/mikioh/tcp v0.0.0-20190314235350-803a9b46060c h1:bzE/A84HN25pxAuk9Eej1Kz9OUelF97nAc82bDquQI8=
+github.com/mikioh/tcp v0.0.0-20190314235350-803a9b46060c/go.mod h1:0SQS9kMwD2VsyFEB++InYyBJroV/FRmBgcydeSUcJms=
+github.com/mikioh/tcpinfo v0.0.0-20190314235526-30a79bb1804b h1:z78hV3sbSMAUoyUMM0I83AUIT6Hu17AWfgjzIbtrYFc=
+github.com/mikioh/tcpinfo v0.0.0-20190314235526-30a79bb1804b/go.mod h1:lxPUiZwKoFL8DUUmalo2yJJUCxbPKtm8OKfqr2/FTNU=
+github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc h1:PTfri+PuQmWDqERdnNMiD9ZejrlswWrCpBEZgWOiTrc=
+github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc/go.mod h1:cGKTAVKx4SxOuR/czcZ/E2RSJ3sfHs8FpHhQ5CWMf9s=
+github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1/go.mod h1:pD8RvIylQ358TN4wwqatJ8rNavkEINozVn9DtGI3dfQ=
+github.com/minio/sha256-simd v0.1.1-0.20190913151208-6de447530771/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
+github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM=
+github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/mr-tron/base58 v1.1.2/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
+github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
+github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
+github.com/multiformats/go-base32 v0.1.0 h1:pVx9xoSPqEIQG8o+UbAe7DNi51oej1NtK+aGkbLYxPE=
+github.com/multiformats/go-base32 v0.1.0/go.mod h1:Kj3tFY6zNr+ABYMqeUNeGvkIC/UYgtWibDcT0rExnbI=
+github.com/multiformats/go-base36 v0.2.0 h1:lFsAbNOGeKtuKozrtBsAkSVhv1p9D0/qedU9rQyccr0=
+github.com/multiformats/go-base36 v0.2.0/go.mod h1:qvnKE++v+2MWCfePClUEjE78Z7P2a1UV0xHgWc0hkp4=
+github.com/multiformats/go-multiaddr v0.1.1/go.mod h1:aMKBKNEYmzmDmxfX88/vz+J5IU55txyt0p4aiWVohjo=
+github.com/multiformats/go-multiaddr v0.14.0 h1:bfrHrJhrRuh/NXH5mCnemjpbGjzRw/b+tJFOD41g2tU=
+github.com/multiformats/go-multiaddr v0.14.0/go.mod h1:6EkVAxtznq2yC3QT5CM1UTAwG0GTP3EWAIcjHuzQ+r4=
+github.com/multiformats/go-multiaddr-dns v0.4.1 h1:whi/uCLbDS3mSEUMb1MsoT4uzUeZB0N32yzufqS0i5M=
+github.com/multiformats/go-multiaddr-dns v0.4.1/go.mod h1:7hfthtB4E4pQwirrz+J0CcDUfbWzTqEzVyYKKIKpgkc=
+github.com/multiformats/go-multiaddr-fmt v0.1.0 h1:WLEFClPycPkp4fnIzoFoV9FVd49/eQsuaL3/CWe167E=
+github.com/multiformats/go-multiaddr-fmt v0.1.0/go.mod h1:hGtDIW4PU4BqJ50gW2quDuPVjyWNZxToGUh/HwTZYJo=
+github.com/multiformats/go-multibase v0.2.0 h1:isdYCVLvksgWlMW9OZRYJEa9pZETFivncJHmHnnd87g=
+github.com/multiformats/go-multibase v0.2.0/go.mod h1:bFBZX4lKCA/2lyOFSAoKH5SS6oPyjtnzK/XTFDPkNuk=
+github.com/multiformats/go-multicodec v0.9.0 h1:pb/dlPnzee/Sxv/j4PmkDRxCOi3hXTz3IbPKOXWJkmg=
+github.com/multiformats/go-multicodec v0.9.0/go.mod h1:L3QTQvMIaVBkXOXXtVmYE+LI16i14xuaojr/H7Ai54k=
+github.com/multiformats/go-multihash v0.0.8/go.mod h1:YSLudS+Pi8NHE7o6tb3D8vrpKa63epEDmG8nTduyAew=
+github.com/multiformats/go-multihash v0.2.3 h1:7Lyc8XfX/IY2jWb/gI7JP+o7JEq9hOa7BFvVU9RSh+U=
+github.com/multiformats/go-multihash v0.2.3/go.mod h1:dXgKXCXjBzdscBLk9JkjINiEsCKRVch90MdaGiKsvSM=
+github.com/multiformats/go-multistream v0.6.0 h1:ZaHKbsL404720283o4c/IHQXiS6gb8qAN5EIJ4PN5EA=
+github.com/multiformats/go-multistream v0.6.0/go.mod h1:MOyoG5otO24cHIg8kf9QW2/NozURlkP/rvi2FQJyCPg=
+github.com/multiformats/go-varint v0.0.7 h1:sWSGR+f/eu5ABZA2ZpYKBILXTTs9JWpdEM/nEGOHFS8=
+github.com/multiformats/go-varint v0.0.7/go.mod h1:r8PUYw/fD/SjBCiKOoDlGF6QawOELpZAu9eioSos/OU=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
+github.com/neelance/astrewrite v0.0.0-20160511093645-99348263ae86/go.mod h1:kHJEU3ofeGjhHklVoIGuVj85JJwZ6kWPaJwCIxgnFmo=
+github.com/neelance/sourcemap v0.0.0-20151028013722-8c68805598ab/go.mod h1:Qr6/a/Q4r9LP1IltGz7tA7iOK1WonHEYhu1HRBA7ZiM=
+github.com/onsi/ginkgo/v2 v2.22.2 h1:/3X8Panh8/WwhU/3Ssa6rCKqPLuAkVY2I0RoyDLySlU=
+github.com/onsi/ginkgo/v2 v2.22.2/go.mod h1:oeMosUL+8LtarXBHu/c0bx2D/K9zyQ6uX3cTyztHwsk=
+github.com/onsi/gomega v1.36.2 h1:koNYke6TVk6ZmnyHrCXba/T/MoLBXFjeC1PtvYgw0A8=
+github.com/onsi/gomega v1.36.2/go.mod h1:DdwyADRjrc825LhMEkD76cHR5+pUnjhUN8GlHlRPHzY=
+github.com/opencontainers/runtime-spec v1.0.2/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0=
+github.com/opencontainers/runtime-spec v1.2.0 h1:z97+pHb3uELt/yiAWD691HNHQIF07bE7dzrbT927iTk=
+github.com/opencontainers/runtime-spec v1.2.0/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0=
+github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8=
+github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 h1:onHthvaw9LFnH4t2DcNVpwGmV9E1BkGknEliJkfwQj0=
+github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58/go.mod h1:DXv8WO4yhMYhSNPKjeNKa5WY9YCIEBRbNzFFPJbWO6Y=
+github.com/pion/datachannel v1.5.10 h1:ly0Q26K1i6ZkGf42W7D4hQYR90pZwzFOjTq5AuCKk4o=
+github.com/pion/datachannel v1.5.10/go.mod h1:p/jJfC9arb29W7WrxyKbepTU20CFgyx5oLo8Rs4Py/M=
+github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s=
+github.com/pion/dtls/v2 v2.2.12 h1:KP7H5/c1EiVAAKUmXyCzPiQe5+bCJrpOeKg/L05dunk=
+github.com/pion/dtls/v2 v2.2.12/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE=
+github.com/pion/dtls/v3 v3.0.4 h1:44CZekewMzfrn9pmGrj5BNnTMDCFwr+6sLH+cCuLM7U=
+github.com/pion/dtls/v3 v3.0.4/go.mod h1:R373CsjxWqNPf6MEkfdy3aSe9niZvL/JaKlGeFphtMg=
+github.com/pion/ice/v2 v2.3.37 h1:ObIdaNDu1rCo7hObhs34YSBcO7fjslJMZV0ux+uZWh0=
+github.com/pion/ice/v2 v2.3.37/go.mod h1:mBF7lnigdqgtB+YHkaY/Y6s6tsyRyo4u4rPGRuOjUBQ=
+github.com/pion/ice/v4 v4.0.6 h1:jmM9HwI9lfetQV/39uD0nY4y++XZNPhvzIPCb8EwxUM=
+github.com/pion/ice/v4 v4.0.6/go.mod h1:y3M18aPhIxLlcO/4dn9X8LzLLSma84cx6emMSu14FGw=
+github.com/pion/interceptor v0.1.37 h1:aRA8Zpab/wE7/c0O3fh1PqY0AJI3fCSEM5lRWJVorwI=
+github.com/pion/interceptor v0.1.37/go.mod h1:JzxbJ4umVTlZAf+/utHzNesY8tmRkM2lVmkS82TTj8Y=
+github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
+github.com/pion/logging v0.2.3 h1:gHuf0zpoh1GW67Nr6Gj4cv5Z9ZscU7g/EaoC/Ke/igI=
+github.com/pion/logging v0.2.3/go.mod h1:z8YfknkquMe1csOrxK5kc+5/ZPAzMxbKLX5aXpbpC90=
+github.com/pion/mdns v0.0.12 h1:CiMYlY+O0azojWDmxdNr7ADGrnZ+V6Ilfner+6mSVK8=
+github.com/pion/mdns v0.0.12/go.mod h1:VExJjv8to/6Wqm1FXK+Ii/Z9tsVk/F5sD/N70cnYFbk=
+github.com/pion/mdns/v2 v2.0.7 h1:c9kM8ewCgjslaAmicYMFQIde2H9/lrZpjBkN8VwoVtM=
+github.com/pion/mdns/v2 v2.0.7/go.mod h1:vAdSYNAT0Jy3Ru0zl2YiW3Rm/fJCwIeM0nToenfOJKA=
+github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
+github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
+github.com/pion/rtcp v1.2.15 h1:LZQi2JbdipLOj4eBjK4wlVoQWfrZbh3Q6eHtWtJBZBo=
+github.com/pion/rtcp v1.2.15/go.mod h1:jlGuAjHMEXwMUHK78RgX0UmEJFV4zUKOFHR7OP+D3D0=
+github.com/pion/rtp v1.8.11 h1:17xjnY5WO5hgO6SD3/NTIUPvSFw/PbLsIJyz1r1yNIk=
+github.com/pion/rtp v1.8.11/go.mod h1:8uMBJj32Pa1wwx8Fuv/AsFhn8jsgw+3rUC2PfoBZ8p4=
+github.com/pion/sctp v1.8.35 h1:qwtKvNK1Wc5tHMIYgTDJhfZk7vATGVHhXbUDfHbYwzA=
+github.com/pion/sctp v1.8.35/go.mod h1:EcXP8zCYVTRy3W9xtOF7wJm1L1aXfKRQzaM33SjQlzg=
+github.com/pion/sdp/v3 v3.0.10 h1:6MChLE/1xYB+CjumMw+gZ9ufp2DPApuVSnDT8t5MIgA=
+github.com/pion/sdp/v3 v3.0.10/go.mod h1:88GMahN5xnScv1hIMTqLdu/cOcUkj6a9ytbncwMCq2E=
+github.com/pion/srtp/v3 v3.0.4 h1:2Z6vDVxzrX3UHEgrUyIGM4rRouoC7v+NiF1IHtp9B5M=
+github.com/pion/srtp/v3 v3.0.4/go.mod h1:1Jx3FwDoxpRaTh1oRV8A/6G1BnFL+QI82eK4ms8EEJQ=
+github.com/pion/stun v0.6.1 h1:8lp6YejULeHBF8NmV8e2787BogQhduZugh5PdhDyyN4=
+github.com/pion/stun v0.6.1/go.mod h1:/hO7APkX4hZKu/D0f2lHzNyvdkTGtIy3NDmLR7kSz/8=
+github.com/pion/stun/v3 v3.0.0 h1:4h1gwhWLWuZWOJIJR9s2ferRO+W3zA/b6ijOI6mKzUw=
+github.com/pion/stun/v3 v3.0.0/go.mod h1:HvCN8txt8mwi4FBvS3EmDghW6aQJ24T+y+1TKjB5jyU=
+github.com/pion/transport/v2 v2.2.1/go.mod h1:cXXWavvCnFF6McHTft3DWS9iic2Mftcz1Aq29pGcU5g=
+github.com/pion/transport/v2 v2.2.4/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLhCCgpRK4p0=
+github.com/pion/transport/v2 v2.2.10 h1:ucLBLE8nuxiHfvkFKnkDQRYWYfp8ejf4YBOPfaQpw6Q=
+github.com/pion/transport/v2 v2.2.10/go.mod h1:sq1kSLWs+cHW9E+2fJP95QudkzbK7wscs8yYgQToO5E=
+github.com/pion/transport/v3 v3.0.1/go.mod h1:UY7kiITrlMv7/IKgd5eTUcaahZx5oUN3l9SzK5f5xE0=
+github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1o0=
+github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo=
+github.com/pion/turn/v2 v2.1.3/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
+github.com/pion/turn/v2 v2.1.6 h1:Xr2niVsiPTB0FPtt+yAWKFUkU1eotQbGgpTIld4x1Gc=
+github.com/pion/turn/v2 v2.1.6/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
+github.com/pion/turn/v4 v4.0.0 h1:qxplo3Rxa9Yg1xXDxxH8xaqcyGUtbHYw4QSCvmFWvhM=
+github.com/pion/turn/v4 v4.0.0/go.mod h1:MuPDkm15nYSklKpN8vWJ9W2M0PlyQZqYt1McGuxG7mA=
+github.com/pion/webrtc/v4 v4.0.8 h1:T1ZmnT9qxIJIt4d8XoiMOBrTClGHDDXNg9e/fh018Qc=
+github.com/pion/webrtc/v4 v4.0.8/go.mod h1:HHBeUVBAC+j4ZFnYhovEFStF02Arb1EyD4G7e7HBTJw=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v0.8.0/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
+github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y=
+github.com/prometheus/client_golang v1.20.5/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
+github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
+github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
+github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
+github.com/prometheus/common v0.0.0-20180801064454-c7de2306084e/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
+github.com/prometheus/common v0.62.0 h1:xasJaQlnWAeyHdUBeGjXmutelfJHWMRr+Fg4QszZ2Io=
+github.com/prometheus/common v0.62.0/go.mod h1:vyBcEuLSvWos9B1+CyL7JZ2up+uFzXhkqml0W5zIY1I=
+github.com/prometheus/procfs v0.0.0-20180725123919-05ee40e3a273/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
+github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
+github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
+github.com/quic-go/qpack v0.5.1 h1:giqksBPnT/HDtZ6VhtFKgoLOWmlyo9Ei6u9PqzIMbhI=
+github.com/quic-go/qpack v0.5.1/go.mod h1:+PC4XFrEskIVkcLzpEkbLqq1uCoxPhQuvK5rH1ZgaEg=
+github.com/quic-go/quic-go v0.49.0 h1:w5iJHXwHxs1QxyBv1EHKuC50GX5to8mJAxvtnttJp94=
+github.com/quic-go/quic-go v0.49.0/go.mod h1:s2wDnmCdooUQBmQfpUSTCYBl1/D4FcqbULMMkASvR6s=
+github.com/quic-go/webtransport-go v0.8.1-0.20241018022711-4ac2c9250e66 h1:4WFk6u3sOT6pLa1kQ50ZVdm8BQFgJNA117cepZxtLIg=
+github.com/quic-go/webtransport-go v0.8.1-0.20241018022711-4ac2c9250e66/go.mod h1:Vp72IJajgeOL6ddqrAhmp7IM9zbTcgkQxD/YdxrVwMw=
+github.com/raulk/go-watchdog v1.3.0 h1:oUmdlHxdkXRJlwfG0O9omj8ukerm8MEQavSiDTEtBsk=
+github.com/raulk/go-watchdog v1.3.0/go.mod h1:fIvOnLbF0b0ZwkB9YU4mOW9Did//4vPZtDqv66NfsMU=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
+github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
+github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
+github.com/shurcooL/component v0.0.0-20170202220835-f88ec8f54cc4/go.mod h1:XhFIlyj5a1fBNx5aJTbKoIq0mNaPvOagO+HjB3EtxrY=
+github.com/shurcooL/events v0.0.0-20181021180414-410e4ca65f48/go.mod h1:5u70Mqkb5O5cxEA8nxTsgrgLehJeAw6Oc4Ab1c/P1HM=
+github.com/shurcooL/github_flavored_markdown v0.0.0-20181002035957-2122de532470/go.mod h1:2dOwnU2uBioM+SGy2aZoq1f/Sd1l9OkAeAUvjSyvgU0=
+github.com/shurcooL/go v0.0.0-20180423040247-9e1955d9fb6e/go.mod h1:TDJrrUr11Vxrven61rcy3hJMUqaf/CLWYhHNPmT14Lk=
+github.com/shurcooL/go-goon v0.0.0-20170922171312-37c2f522c041/go.mod h1:N5mDOmsrJOB+vfqUK+7DmDyjhSLIIBnXo9lvZJj3MWQ=
+github.com/shurcooL/gofontwoff v0.0.0-20180329035133-29b52fc0a18d/go.mod h1:05UtEgK5zq39gLST6uB0cf3NEHjETfB4Fgr3Gx5R9Vw=
+github.com/shurcooL/gopherjslib v0.0.0-20160914041154-feb6d3990c2c/go.mod h1:8d3azKNyqcHP1GaQE/c6dDgjkgSx2BZ4IoEi4F1reUI=
+github.com/shurcooL/highlight_diff v0.0.0-20170515013008-09bb4053de1b/go.mod h1:ZpfEhSmds4ytuByIcDnOLkTHGUI6KNqRNPDLHDk+mUU=
+github.com/shurcooL/highlight_go v0.0.0-20181028180052-98c3abbbae20/go.mod h1:UDKB5a1T23gOMUJrI+uSuH0VRDStOiUVSjBTRDVBVag=
+github.com/shurcooL/home v0.0.0-20181020052607-80b7ffcb30f9/go.mod h1:+rgNQw2P9ARFAs37qieuu7ohDNQ3gds9msbT2yn85sg=
+github.com/shurcooL/htmlg v0.0.0-20170918183704-d01228ac9e50/go.mod h1:zPn1wHpTIePGnXSHpsVPWEktKXHr6+SS6x/IKRb7cpw=
+github.com/shurcooL/httperror v0.0.0-20170206035902-86b7830d14cc/go.mod h1:aYMfkZ6DWSJPJ6c4Wwz3QtW22G7mf/PEgaB9k/ik5+Y=
+github.com/shurcooL/httpfs v0.0.0-20171119174359-809beceb2371/go.mod h1:ZY1cvUeJuFPAdZ/B6v7RHavJWZn2YPVFQ1OSXhCGOkg=
+github.com/shurcooL/httpgzip v0.0.0-20180522190206-b1c53ac65af9/go.mod h1:919LwcH0M7/W4fcZ0/jy0qGght1GIhqyS/EgWGH2j5Q=
+github.com/shurcooL/issues v0.0.0-20181008053335-6292fdc1e191/go.mod h1:e2qWDig5bLteJ4fwvDAc2NHzqFEthkqn7aOZAOpj+PQ=
+github.com/shurcooL/issuesapp v0.0.0-20180602232740-048589ce2241/go.mod h1:NPpHK2TI7iSaM0buivtFUc9offApnI0Alt/K8hcHy0I=
+github.com/shurcooL/notifications v0.0.0-20181007000457-627ab5aea122/go.mod h1:b5uSkrEVM1jQUspwbixRBhaIjIzL2xazXp6kntxYle0=
+github.com/shurcooL/octicon v0.0.0-20181028054416-fa4f57f9efb2/go.mod h1:eWdoE5JD4R5UVWDucdOPg1g2fqQRq78IQa9zlOV1vpQ=
+github.com/shurcooL/reactions v0.0.0-20181006231557-f2e0b4ca5b82/go.mod h1:TCR1lToEk4d2s07G3XGfz2QrgHXg4RJBvjrOozvoWfk=
+github.com/shurcooL/sanitized_anchor_name v0.0.0-20170918181015-86672fcb3f95/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
+github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
+github.com/shurcooL/users v0.0.0-20180125191416-49c67e49c537/go.mod h1:QJTqeLYEDaXHZDBsXlPCDqdhQuJkuw4NOtaxYe3xii4=
+github.com/shurcooL/webdavfs v0.0.0-20170829043945-18c3829fa133/go.mod h1:hKmq5kWdCj2z2KEozexVbfEZIWiTjhE0+UjmZgPqehw=
+github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
+github.com/sourcegraph/annotate v0.0.0-20160123013949-f4cad6c6324d/go.mod h1:UdhH50NIW0fCiwBSr0co2m7BnFLdv4fQTgdqdJTHFeE=
+github.com/sourcegraph/syntaxhighlight v0.0.0-20170531221838-bd320f5d308e/go.mod h1:HuIsMU8RRBOtsCgI77wP899iHVBQpCmg4ErYMZB+2IA=
+github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
+github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07/go.mod h1:kDXzergiv9cbyO7IOYJZWg1U88JhDg3PB6klq9Hg2pA=
+github.com/urfave/cli v1.22.2/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
+github.com/viant/assertly v0.4.8/go.mod h1:aGifi++jvCrUaklKEKT0BU95igDNaqkvz+49uaYMPRU=
+github.com/viant/toolbox v0.24.0/go.mod h1:OxMCG57V0PXuIP2HNQrtJf2CjqdmbrOx5EkMILuUhzM=
+github.com/wlynxg/anet v0.0.3/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
+github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=
+github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
+github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+go.opencensus.io v0.18.0/go.mod h1:vKdFvxhtzZ9onBp9VKHK8z/sRpBMnKAsufL7wlDrCOA=
+go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/dig v1.18.0 h1:imUL1UiY0Mg4bqbFfsRQO5G4CGRBec/ZujWTvSVp3pw=
+go.uber.org/dig v1.18.0/go.mod h1:Us0rSJiThwCv2GteUN0Q7OKvU7n5J4dxZ9JKUXozFdE=
+go.uber.org/fx v1.23.0 h1:lIr/gYWQGfTwGcSXWXu4vP5Ws6iqnNEIY+F/aFzCKTg=
+go.uber.org/fx v1.23.0/go.mod h1:o/D9n+2mLP6v1EG+qsdT1O8wKopYAsqZasju97SDFCU=
+go.uber.org/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.uber.org/mock v0.5.0 h1:KAMbZvZPyBPWgD14IrIQ38QCyjwpvVVV6K/bHl1IwQU=
+go.uber.org/mock v0.5.0/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
+go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
+go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
+go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
+go.uber.org/zap v1.19.1/go.mod h1:j3DNczoxDZroyBnOT1L/Q79cfUMGZxlv/9dzN7SM1rI=
+go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
+go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+go4.org v0.0.0-20180809161055-417644f6feb5/go.mod h1:MkTOUMDaeVYJUOUsaDXIhWPZYa1yOyC1qaOBpL57BhE=
+golang.org/x/build v0.0.0-20190111050920-041ab4dc3f9d/go.mod h1:OWs+y06UdEOHN4y+MfF/py+xQ/tYqIWW03b70/CG9Rw=
+golang.org/x/crypto v0.0.0-20181030102418-4d3f4d9ffa16/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20190313024323-a1f597ede03a/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20200602180216-279210d13fed/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE=
+golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98yw=
+golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
+golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
+golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
+golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
+golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c h1:KL/ZBHXgKGVmuZBZ01Lt57yE5ws8ZPSkkihmEyq7FXc=
+golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c/go.mod h1:tujkw807nyEEAamNbDrEGzRav+ilXA7PCRAd6xsmwiU=
+golang.org/x/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
+golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
+golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
+golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
+golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
+golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/mod v0.23.0 h1:Zb7khfcRGKk+kqfxFaP5tZqCnDZMjC5VtUBs87Hr6QM=
+golang.org/x/mod v0.23.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY=
+golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20181029044818-c44066c5c816/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20181106065722-10aee1819953/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190313220215-9f648a60d977/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
+golang.org/x/net v0.0.0-20210423184538-5f58ad60dda6/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
+golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
+golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
+golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
+golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
+golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
+golang.org/x/oauth2 v0.0.0-20181017192945-9dcd33a902f4/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
+golang.org/x/oauth2 v0.0.0-20181203162652-d668ce993890/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
+golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
+golang.org/x/perf v0.0.0-20180704124530-6e6d33e29852/go.mod h1:JLpeXjPJfIyPr5TlbXLkXWLhP8nz10XfvxElABhCtcw=
+golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
+golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sys v0.0.0-20180810173357-98c5dad5d1a0/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20181029174526-d69651ed3497/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190316082340-a2f829d7f35f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200124204421-9fbb57f87de9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200602225109-6fdc65e7d980/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210303074136-134d130e1a04/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210426080607-c94f62235c83/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/term v0.7.0/go.mod h1:P32HKFT3hSsZrRxla30E9HqToFYAQPCMs/zFMBUFqPY=
+golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
+golang.org/x/term v0.11.0/go.mod h1:zC9APTIj3jG3FdV/Ons+XE1riIZXG4aZ4GTHiPZJPIU=
+golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
+golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
+golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
+golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20181030000716-a0a13e073c7b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20181030221726-6c7e314b6563/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
+golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
+golang.org/x/tools v0.29.0 h1:Xx0h3TtM9rzQpQuR4dKLrdglAmCEN5Oi+P74JdhdzXE=
+golang.org/x/tools v0.29.0/go.mod h1:KMQVMRsVxU6nHCFXrBPhDB8XncLNLM0lIy/F14RP588=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/api v0.0.0-20180910000450-7ca32eb868bf/go.mod h1:4mhQ8q/RsB7i+udVvVy5NUi08OU8ZlA0gRVgrF7VFY0=
+google.golang.org/api v0.0.0-20181030000543-1d582fd0359e/go.mod h1:4mhQ8q/RsB7i+udVvVy5NUi08OU8ZlA0gRVgrF7VFY0=
+google.golang.org/api v0.1.0/go.mod h1:UGEZY7KEX120AnNLIHFMKIo4obdJhkp2tPbaPlQx13Y=
+google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
+google.golang.org/appengine v1.2.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
+google.golang.org/appengine v1.3.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
+google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
+google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
+google.golang.org/genproto v0.0.0-20180831171423-11092d34479b/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
+google.golang.org/genproto v0.0.0-20181029155118-b69ba1387ce2/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
+google.golang.org/genproto v0.0.0-20181202183823-bd91e49a0898/go.mod h1:7Ep/1NZk928CDR8SjdVbjWNpdIf6nzjE3BTgJDr2Atg=
+google.golang.org/genproto v0.0.0-20190306203927-b5d61aea6440/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
+google.golang.org/grpc v1.14.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
+google.golang.org/grpc v1.16.0/go.mod h1:0JHn/cJsOMiMfNA9+DeHDlAU7KAAB5GDlYFpa9MZMio=
+google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3cWCs=
+google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
+google.golang.org/protobuf v1.36.4 h1:6A3ZDJHn/eNqc1i+IdefRzy/9PokBTPvcqMySR7NNIM=
+google.golang.org/protobuf v1.36.4/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
+gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+grpc.go4.org v0.0.0-20170609214715-11d0a25b4919/go.mod h1:77eQGdRu53HpSqPFJFmuJdjuHRquDANNeA4x7B8WQ9o=
+honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
+honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
+honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
+lukechampine.com/blake3 v1.3.0 h1:sJ3XhFINmHSrYCgl958hscfIa3bw8x4DqMP3u1YvoYE=
+lukechampine.com/blake3 v1.3.0/go.mod h1:0OFRp7fBtAylGVCO40o87sbupkyIGgbpv1+M1k1LM6k=
+sourcegraph.com/sourcegraph/go-diff v0.5.0/go.mod h1:kuch7UrkMzY0X+p9CRK03kfuPQ2zzQcaEFbx8wA8rck=
+sourcegraph.com/sqs/pbtypes v0.0.0-20180604144634-d3ebe8f20ae4/go.mod h1:ketZ/q3QxT9HOBeFhu6RdvsftgpsbFHBF5Cas6cDKZ0=
diff --git a/networking/forwarder/main.go b/networking/forwarder/main.go
new file mode 100644
index 00000000..dd2a9ea4
--- /dev/null
+++ b/networking/forwarder/main.go
@@ -0,0 +1,59 @@
+package main
+
+import (
+ "context"
+ "flag"
+ forwarder "forwarder/src"
+ "log"
+ "os"
+ "os/signal"
+ "syscall"
+)
+
+var nodeID = flag.String("node-id", "", "Node ID (defaults to FORWARDER_NODE_ID env var or a new UUID)")
+
+func main() {
+ flag.Parse()
+
+ id := *nodeID
+ if id != "" {
+ forwarder.SetNodeId(id)
+ } else {
+ id = forwarder.GetNodeId()
+ }
+ log.Printf("Starting forwarder with node ID: %s", id)
+
+ args := flag.Args()
+ if len(args) == 0 {
+ log.Fatal("forwarding pairs argument is required as the first positional argument (of the form {source}|{sink}) where source and sink sqlite:db_file:table_name or libp2p:topic")
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ forwardingPairs := args[0]
+ connections, err := forwarder.ParseForwardingPairs(forwardingPairs, ctx, cancel)
+ if err != nil {
+ log.Fatalf("Failed to parse forwarding pairs: %v", err)
+ }
+ for _, conn := range connections {
+ log.Printf("Forwarding Pair %v", conn)
+ }
+
+ for _, conn := range connections {
+ fwd, err := forwarder.NewForwarder(conn)
+ if err != nil {
+ log.Fatalf("Failed to create forwarder: %v", err)
+ }
+ fwd.Start(ctx)
+ }
+ sig := make(chan os.Signal, 1)
+ signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
+ go func() {
+ <-sig
+ cancel()
+ }()
+
+ <-ctx.Done()
+ log.Println("Forwarder is shutting down...")
+}
diff --git a/networking/forwarder/src/config.go b/networking/forwarder/src/config.go
new file mode 100644
index 00000000..ad0a392e
--- /dev/null
+++ b/networking/forwarder/src/config.go
@@ -0,0 +1,91 @@
+package forwarder
+
+import (
+ "context"
+ "fmt"
+ "strings"
+)
+
+func ParseForwardingPairs(pairsStr string, ctx context.Context, cancel context.CancelFunc) ([]ForwardingPair, error) {
+ if pairsStr == "" {
+ return nil, fmt.Errorf("forwarding pairs string is empty")
+ }
+
+ pairStrs := strings.Split(pairsStr, ",")
+ var connections []ForwardingPair
+
+ for _, pairStr := range pairStrs {
+ pairStr = strings.TrimSpace(pairStr)
+ if pairStr == "" {
+ continue
+ }
+
+ parts := strings.Split(pairStr, "|")
+ if len(parts) != 2 {
+ return nil, fmt.Errorf("invalid forwarding pair format: %s", pairStr)
+ }
+
+ sourceStr := strings.TrimSpace(parts[0])
+ sinkStr := strings.TrimSpace(parts[1])
+
+ sourceType := strings.Split(sourceStr, ":")[0]
+ sinkType := strings.Split(sinkStr, ":")[0]
+ if sinkType == sourceType {
+ return nil, fmt.Errorf("source and sink types cannot be the same: %s", pairStr)
+ }
+
+ sourceConn, err := parseEndpoint(sourceStr, ctx, cancel)
+ if err != nil {
+ return nil, fmt.Errorf("invalid source endpoint '%s': %w", sourceStr, err)
+ }
+
+ sinkConn, err := parseEndpoint(sinkStr, ctx, cancel)
+ if err != nil {
+ return nil, fmt.Errorf("invalid sink endpoint '%s': %w", sinkStr, err)
+ }
+
+ conn := ForwardingPair{
+ source: sourceConn,
+ sink: sinkConn,
+ }
+ connections = append(connections, conn)
+ }
+ tables := make(map[string]bool)
+ for _, conn := range connections {
+ if conn.sink.getType() == "sqlite" {
+ tableName := conn.sink.(*sqliteConnector).tableName
+ if _, ok := tables[tableName]; ok {
+ return nil, fmt.Errorf("sink table '%s' already used in another connection", tableName)
+ }
+ tables[tableName] = true
+ }
+ }
+
+ return connections, nil
+}
+
+func parseEndpoint(endpointStr string, ctx context.Context, cancel context.CancelFunc) (connection, error) {
+ parts := strings.SplitN(endpointStr, ":", 2)
+ if len(parts) < 2 || parts[1] == "" {
+ return nil, fmt.Errorf("invalid endpoint format: %s", endpointStr)
+ }
+
+ endpointType := parts[0]
+ endpointArgsStr := parts[1]
+
+ switch endpointType {
+ case "sqlite":
+ args := strings.SplitN(endpointArgsStr, ":", 2)
+ if len(args) != 2 || args[0] == "" || args[1] == "" {
+ return nil, fmt.Errorf("invalid sqlite endpoint format: %s. Expected 'sqlite:db_file:table'", endpointStr)
+ }
+ return newSQLiteConnector(args[0], args[1])
+ case "libp2p":
+ if strings.Contains(endpointArgsStr, ":") {
+ return nil, fmt.Errorf("invalid libp2p topic format: %s. Topic should not contain ':'", endpointStr)
+ }
+ return newLibP2PConnector(endpointArgsStr, ctx, cancel), nil
+ default:
+ return nil, fmt.Errorf("unknown endpoint type: %s", endpointType)
+ }
+}
diff --git a/networking/forwarder/src/forwarder.go b/networking/forwarder/src/forwarder.go
new file mode 100644
index 00000000..8ad32b35
--- /dev/null
+++ b/networking/forwarder/src/forwarder.go
@@ -0,0 +1,133 @@
+package forwarder
+
+import (
+ "context"
+ "fmt"
+ "log"
+ "time"
+)
+
+type libP2PToSqliteForwarder struct {
+ source LibP2PConnection
+ sink SQLiteConnection
+ recordStore stateStoreInterface
+}
+
+func newLibP2PToSqliteForwarder(source LibP2PConnection, sink SQLiteConnection) (*libP2PToSqliteForwarder, error) {
+ latestRowIds, err := sink.getLatestRowIds()
+ if err != nil {
+ return nil, fmt.Errorf("failed to get latest row IDs: %w", err)
+ }
+ return &libP2PToSqliteForwarder{
+ source: source,
+ sink: sink,
+ recordStore: newStateStore(latestRowIds),
+ }, nil
+}
+
+func (f *libP2PToSqliteForwarder) Start(ctx context.Context) error {
+ f.source.tail(func(record RecordData) error {
+ f.recordStore.onRecord(record)
+ return nil
+ })
+
+ go func() {
+ ticker := time.NewTicker(10 * time.Millisecond)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ msgs := f.recordStore.getWriteableMessages()
+ for _, msg := range msgs {
+ if err := f.sink.write(msg); err != nil {
+ log.Printf("Error writing to sink: %v", err)
+ }
+ }
+ }
+ }
+ }()
+
+ // Resend handler with less frequent checks
+ go func() {
+ ticker := time.NewTicker(500 * time.Millisecond) // Less frequent than before
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ reqs := f.recordStore.getResendRequests()
+ for _, req := range reqs {
+ if err := f.source.writeResend(req); err != nil {
+ log.Printf("Error writing resend request: %v", err)
+ }
+ }
+ }
+ }
+ }()
+
+ return nil
+}
+
+type sqliteToLibP2PForwarder struct {
+ source SQLiteConnection
+ sink LibP2PConnection
+}
+
+func newSqliteToLibP2PForwarder(source SQLiteConnection, sink LibP2PConnection) (*sqliteToLibP2PForwarder, error) {
+ return &sqliteToLibP2PForwarder{
+ source: source,
+ sink: sink,
+ }, nil
+}
+
+func (f *sqliteToLibP2PForwarder) Start(ctx context.Context) error {
+ // Handle resend requests
+ f.sink.tailResend(func(req ResendRequest) error {
+ if req.SourceNodeID != f.source.getNodeId() {
+ return nil
+ }
+ if req.SourcePath != f.source.getTablePath() {
+ return nil
+ }
+
+ // Process resends in a separate goroutine to not block
+ go func() {
+ for _, gap := range req.Gaps {
+ records, err := f.source.readRange(gap.Start, gap.End)
+ if err != nil {
+ log.Printf("Error getting records for resend: %v", err)
+ continue
+ }
+ // Send resend records - libp2p connector will handle batching
+ for _, rec := range records {
+ if err := f.sink.write(rec); err != nil {
+ log.Printf("Error writing resend record: %v", err)
+ }
+ }
+ }
+ }()
+ return nil
+ })
+
+ // Tail new records - libp2p connector handles async batching internally
+ f.source.tail(func(record RecordData) error {
+ if err := f.sink.write(record); err != nil {
+ log.Printf("Error writing record: %v", err)
+ }
+ return nil
+ })
+
+ return nil
+}
+
+func NewForwarder(forwardingPair ForwardingPair) (Forwarder, error) {
+ if forwardingPair.source.getType() == "libp2p" && forwardingPair.sink.getType() == "sqlite" {
+ return newLibP2PToSqliteForwarder(forwardingPair.source.(*libP2PConnector), forwardingPair.sink.(*sqliteConnector))
+ } else if forwardingPair.source.getType() == "sqlite" && forwardingPair.sink.getType() == "libp2p" {
+ return newSqliteToLibP2PForwarder(forwardingPair.source.(*sqliteConnector), forwardingPair.sink.(*libP2PConnector))
+ }
+ return nil, fmt.Errorf("unsupported forwarding pair: %v", forwardingPair)
+}
diff --git a/networking/forwarder/src/forwarder_test.go b/networking/forwarder/src/forwarder_test.go
new file mode 100644
index 00000000..82d78952
--- /dev/null
+++ b/networking/forwarder/src/forwarder_test.go
@@ -0,0 +1,474 @@
+package forwarder
+
+import (
+ "context"
+ "fmt"
+ "reflect"
+ "testing"
+ "time"
+)
+
+type mockLibP2PConnector struct {
+ tailHandler func(RecordData) error
+ tailResendHandler func(ResendRequest) error
+ writtenRecords []RecordData
+ writeErr error
+ resendRequests []ResendRequest
+ writeResendErr error
+}
+
+func (m *mockLibP2PConnector) tail(handler func(record RecordData) error) {
+ m.tailHandler = handler
+}
+
+func (m *mockLibP2PConnector) tailResend(handler func(req ResendRequest) error) {
+ m.tailResendHandler = handler
+}
+
+func (m *mockLibP2PConnector) write(record RecordData) error {
+ m.writtenRecords = append(m.writtenRecords, record)
+ return m.writeErr
+}
+
+func (m *mockLibP2PConnector) writeResend(req ResendRequest) error {
+ m.resendRequests = append(m.resendRequests, req)
+ return m.writeResendErr
+}
+
+func (m *mockLibP2PConnector) close() error {
+ return nil
+}
+
+func (m *mockLibP2PConnector) getType() string {
+ return "libp2p"
+}
+
+func (m *mockLibP2PConnector) SendRecord(record RecordData) error {
+ if m.tailHandler == nil {
+ return fmt.Errorf("no tail handler registered")
+ }
+ return m.tailHandler(record)
+}
+
+func (m *mockLibP2PConnector) SendResend(req ResendRequest) error {
+ if m.tailResendHandler == nil {
+ return fmt.Errorf("no tailResend handler registered")
+ }
+ return m.tailResendHandler(req)
+}
+
+type mockSqliteConnector struct {
+ getLatestRowIdsRet map[SourceKey]int64
+ getLatestRowIdsErr error
+ writtenRecords []RecordData
+ writeErr error
+ readRangeCalls []struct{ start, end int64 }
+ readRangeRet []RecordData
+ readRangeErr error
+ nodeId string
+ tablePath string
+ tailHandler func(RecordData) error
+}
+
+func (m *mockSqliteConnector) getLatestRowIds() (map[SourceKey]int64, error) {
+ return m.getLatestRowIdsRet, m.getLatestRowIdsErr
+}
+
+func (m *mockSqliteConnector) write(record RecordData) error {
+ m.writtenRecords = append(m.writtenRecords, record)
+ return m.writeErr
+}
+
+func (m *mockSqliteConnector) readRange(start, end int64) ([]RecordData, error) {
+ m.readRangeCalls = append(m.readRangeCalls, struct{ start, end int64 }{start, end})
+ return m.readRangeRet, m.readRangeErr
+}
+
+func (m *mockSqliteConnector) tail(handler func(record RecordData) error) {
+ m.tailHandler = handler
+}
+
+func (m *mockSqliteConnector) close() error {
+ return nil
+}
+
+func (m *mockSqliteConnector) getType() string {
+ return "sqlite"
+}
+
+func (m *mockSqliteConnector) SendRecord(record RecordData) error {
+ if m.tailHandler == nil {
+ return fmt.Errorf("no tail handler registered")
+ }
+ return m.tailHandler(record)
+}
+
+func (m *mockSqliteConnector) getNodeId() string {
+ return m.nodeId
+}
+
+func (m *mockSqliteConnector) getTablePath() string {
+ return m.tablePath
+}
+
+func TestNewLibP2PToSqliteForwarder(t *testing.T) {
+ source := &mockLibP2PConnector{}
+ sink := &mockSqliteConnector{
+ getLatestRowIdsRet: map[SourceKey]int64{},
+ }
+ f, err := newLibP2PToSqliteForwarder(source, sink)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if f == nil {
+ t.Fatal("expected non-nil forwarder")
+ }
+}
+
+func TestLibP2PToSqliteForwarder_Start_InOrderRecords(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ source := &mockLibP2PConnector{}
+ sink := &mockSqliteConnector{
+ getLatestRowIdsRet: map[SourceKey]int64{},
+ }
+
+ f, err := newLibP2PToSqliteForwarder(source, sink)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ err = f.Start(ctx)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ key := SourceKey{SourceNodeId: "node1", SourcePath: "path1"}
+
+ rec1 := RecordData{TrackingData: TrackingData{SourceKey: key, SourceRowID: 1}}
+ source.SendRecord(rec1)
+
+ time.Sleep(500 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 1 {
+ t.Fatalf("expected 1 written record, got %d", len(sink.writtenRecords))
+ }
+ if !reflect.DeepEqual(sink.writtenRecords[0], rec1) {
+ t.Fatal("written record mismatch")
+ }
+
+ rec2 := RecordData{TrackingData: TrackingData{SourceKey: key, SourceRowID: 2}}
+ source.SendRecord(rec2)
+
+ time.Sleep(200 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 2 {
+ t.Fatalf("expected 2 written records, got %d", len(sink.writtenRecords))
+ }
+ if !reflect.DeepEqual(sink.writtenRecords[1], rec2) {
+ t.Fatal("written record mismatch")
+ }
+}
+
+func TestLibP2PToSqliteForwarder_Start_OutOfOrderRecords(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ source := &mockLibP2PConnector{}
+ sink := &mockSqliteConnector{
+ getLatestRowIdsRet: map[SourceKey]int64{},
+ }
+
+ f, err := newLibP2PToSqliteForwarder(source, sink)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ err = f.Start(ctx)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ key := SourceKey{SourceNodeId: "node1", SourcePath: "path1"}
+
+ rec1 := RecordData{TrackingData: TrackingData{SourceKey: key, SourceRowID: 1}}
+ source.SendRecord(rec1)
+
+ time.Sleep(200 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 1 {
+ t.Fatalf("expected 1 written record, got %d", len(sink.writtenRecords))
+ }
+
+ rec3 := RecordData{TrackingData: TrackingData{SourceKey: key, SourceRowID: 3}}
+ source.SendRecord(rec3)
+
+ time.Sleep(200 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 1 {
+ t.Fatalf("expected still 1 written record, got %d", len(sink.writtenRecords))
+ }
+
+ time.Sleep(5500 * time.Millisecond) // Wait for resend ticker
+
+ if len(source.resendRequests) != 1 {
+ t.Fatalf("expected 1 resend request, got %d", len(source.resendRequests))
+ }
+
+ req := source.resendRequests[0]
+ if req.SourceNodeID != "node1" || req.SourcePath != "path1" {
+ t.Fatal("resend request mismatch")
+ }
+ if len(req.Gaps) != 1 || req.Gaps[0].Start != 2 || req.Gaps[0].End != 2 {
+ t.Fatal("gap mismatch")
+ }
+
+ rec2 := RecordData{TrackingData: TrackingData{SourceKey: key, SourceRowID: 2}}
+ source.SendRecord(rec2)
+
+ time.Sleep(200 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 3 {
+ t.Fatalf("expected 3 written records, got %d", len(sink.writtenRecords))
+ }
+ // Check order: rec1, rec2, rec3
+ if !reflect.DeepEqual(sink.writtenRecords[1], rec2) || !reflect.DeepEqual(sink.writtenRecords[2], rec3) {
+ t.Fatal("written records order mismatch")
+ }
+}
+
+func TestLibP2PToSqliteForwarder_Start_MultipleSources(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ source := &mockLibP2PConnector{}
+ sink := &mockSqliteConnector{
+ getLatestRowIdsRet: map[SourceKey]int64{},
+ }
+
+ f, err := newLibP2PToSqliteForwarder(source, sink)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ err = f.Start(ctx)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ key1 := SourceKey{SourceNodeId: "node1", SourcePath: "path1"}
+ key2 := SourceKey{SourceNodeId: "node2", SourcePath: "path2"}
+
+ rec1_1 := RecordData{TrackingData: TrackingData{SourceKey: key1, SourceRowID: 1}}
+ source.SendRecord(rec1_1)
+
+ rec2_1 := RecordData{TrackingData: TrackingData{SourceKey: key2, SourceRowID: 1}}
+ source.SendRecord(rec2_1)
+
+ time.Sleep(200 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 2 {
+ t.Fatalf("expected 2 written records, got %d", len(sink.writtenRecords))
+ }
+
+ rec1_3 := RecordData{TrackingData: TrackingData{SourceKey: key1, SourceRowID: 3}}
+ source.SendRecord(rec1_3)
+
+ time.Sleep(200 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 2 {
+ t.Fatalf("expected still 2 written records, got %d", len(sink.writtenRecords))
+ }
+
+ time.Sleep(5500 * time.Millisecond)
+
+ if len(source.resendRequests) != 1 {
+ t.Fatalf("expected 1 resend request, got %d", len(source.resendRequests))
+ }
+ if source.resendRequests[0].SourceNodeID != "node1" {
+ t.Fatal("resend for wrong source")
+ }
+}
+
+func TestLibP2PToSqliteForwarder_Start_WithInitialLatest(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ key := SourceKey{SourceNodeId: "node1", SourcePath: "path1"}
+
+ source := &mockLibP2PConnector{}
+ sink := &mockSqliteConnector{
+ getLatestRowIdsRet: map[SourceKey]int64{key: 5},
+ }
+
+ f, err := newLibP2PToSqliteForwarder(source, sink)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ err = f.Start(ctx)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ rec6 := RecordData{TrackingData: TrackingData{SourceKey: key, SourceRowID: 6}}
+ source.SendRecord(rec6)
+
+ time.Sleep(200 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 1 {
+ t.Fatalf("expected 1 written record, got %d", len(sink.writtenRecords))
+ }
+
+ rec7 := RecordData{TrackingData: TrackingData{SourceKey: key, SourceRowID: 7}}
+ source.SendRecord(rec7)
+
+ time.Sleep(200 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 2 {
+ t.Fatalf("expected 2 written records, got %d", len(sink.writtenRecords))
+ }
+}
+
+func TestNewSqliteToLibP2PForwarder(t *testing.T) {
+ source := &mockSqliteConnector{}
+ sink := &mockLibP2PConnector{}
+ f, err := newSqliteToLibP2PForwarder(source, sink)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if f == nil {
+ t.Fatal("expected non-nil forwarder")
+ }
+}
+
+func TestSqliteToLibP2PForwarder_Start_TailRecords(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ source := &mockSqliteConnector{
+ nodeId: "node1",
+ tablePath: "path1",
+ }
+ sink := &mockLibP2PConnector{}
+
+ f, err := newSqliteToLibP2PForwarder(source, sink)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ err = f.Start(ctx)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ rec1 := RecordData{TrackingData: TrackingData{SourceRowID: 1}}
+ source.SendRecord(rec1)
+
+ time.Sleep(100 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 1 {
+ t.Fatalf("expected 1 written record, got %d", len(sink.writtenRecords))
+ }
+ if !reflect.DeepEqual(sink.writtenRecords[0], rec1) {
+ t.Fatal("written record mismatch")
+ }
+
+ rec2 := RecordData{TrackingData: TrackingData{SourceRowID: 2}}
+ source.SendRecord(rec2)
+
+ time.Sleep(100 * time.Millisecond)
+
+ if len(sink.writtenRecords) != 2 {
+ t.Fatalf("expected 2 written records, got %d", len(sink.writtenRecords))
+ }
+ if !reflect.DeepEqual(sink.writtenRecords[1], rec2) {
+ t.Fatal("written record mismatch")
+ }
+}
+
+func TestSqliteToLibP2PForwarder_Start_ResendRequest_Matching(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ source := &mockSqliteConnector{
+ nodeId: "node1",
+ tablePath: "path1",
+ readRangeRet: []RecordData{
+ {TrackingData: TrackingData{SourceRowID: 5}},
+ },
+ }
+ sink := &mockLibP2PConnector{}
+
+ f, err := newSqliteToLibP2PForwarder(source, sink)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ err = f.Start(ctx)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ req := ResendRequest{
+ SourceNodeID: "node1",
+ SourcePath: "path1",
+ Gaps: []GapRange{{Start: 5, End: 6}},
+ }
+ sink.SendResend(req)
+
+ time.Sleep(100 * time.Millisecond)
+
+ if len(source.readRangeCalls) != 1 {
+ t.Fatalf("expected 1 readRange call, got %d", len(source.readRangeCalls))
+ }
+ if source.readRangeCalls[0].start != 5 || source.readRangeCalls[0].end != 6 {
+ t.Fatal("readRange args mismatch")
+ }
+
+ if len(sink.writtenRecords) != 1 {
+ t.Fatalf("expected 1 written record from resend, got %d", len(sink.writtenRecords))
+ }
+ if sink.writtenRecords[0].SourceRowID != 5 {
+ t.Fatal("resend record mismatch")
+ }
+}
+
+func TestSqliteToLibP2PForwarder_Start_ResendRequest_NotMatching(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ source := &mockSqliteConnector{
+ nodeId: "node1",
+ tablePath: "path1",
+ }
+ sink := &mockLibP2PConnector{}
+
+ f, err := newSqliteToLibP2PForwarder(source, sink)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ err = f.Start(ctx)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ req := ResendRequest{
+ SourceNodeID: "node2",
+ SourcePath: "path2",
+ Gaps: []GapRange{{Start: 5, End: 5}},
+ }
+ sink.SendResend(req)
+
+ time.Sleep(100 * time.Millisecond)
+
+ if len(source.readRangeCalls) != 0 {
+ t.Fatalf("expected 0 readRange calls, got %d", len(source.readRangeCalls))
+ }
+
+ if len(sink.writtenRecords) != 0 {
+ t.Fatalf("expected 0 written records, got %d", len(sink.writtenRecords))
+ }
+}
diff --git a/networking/forwarder/src/identity.go b/networking/forwarder/src/identity.go
new file mode 100644
index 00000000..5bf32351
--- /dev/null
+++ b/networking/forwarder/src/identity.go
@@ -0,0 +1,29 @@
+package forwarder
+
+import (
+ "os"
+ "sync"
+
+ "github.com/google/uuid"
+)
+
+var (
+ generatedNodeID string
+ nodeIDOnce sync.Once
+)
+
+func GetNodeId() string {
+ if id := os.Getenv("FORWARDER_NODE_ID"); id != "" {
+ return id
+ }
+
+ nodeIDOnce.Do(func() {
+ generatedNodeID = uuid.New().String()
+ })
+
+ return generatedNodeID
+}
+
+func SetNodeId(id string) {
+ os.Setenv("FORWARDER_NODE_ID", id)
+}
diff --git a/networking/forwarder/src/libp2p.go b/networking/forwarder/src/libp2p.go
new file mode 100644
index 00000000..584e2b04
--- /dev/null
+++ b/networking/forwarder/src/libp2p.go
@@ -0,0 +1,414 @@
+package forwarder
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "encoding/json"
+ "log"
+ "sync"
+ "time"
+
+ "github.com/libp2p/go-libp2p"
+ pubsub "github.com/libp2p/go-libp2p-pubsub"
+ "github.com/libp2p/go-libp2p/core/crypto"
+ "github.com/libp2p/go-libp2p/core/host"
+ "github.com/libp2p/go-libp2p/core/network"
+ "github.com/libp2p/go-libp2p/core/peer"
+ "github.com/libp2p/go-libp2p/core/pnet"
+ mdns "github.com/libp2p/go-libp2p/p2p/discovery/mdns"
+ "github.com/libp2p/go-libp2p/p2p/security/noise"
+)
+
+var node host.Host
+var ps *pubsub.PubSub
+var mdnsSer mdns.Service
+var once sync.Once
+var mu sync.Mutex
+var refCount int
+var topicsMap = make(map[string]*pubsub.Topic)
+
+type discoveryNotifee struct {
+ h host.Host
+}
+
+func (n *discoveryNotifee) HandlePeerFound(pi peer.AddrInfo) {
+ if n.h.ID() >= pi.ID {
+ return
+ }
+ if n.h.Network().Connectedness(pi.ID) == network.Connected {
+ return
+ }
+ ctx := context.Background()
+ if err := n.h.Connect(ctx, pi); err != nil {
+ log.Printf("Failed to connect to %s: %v", pi.ID.String(), err)
+ } else {
+ log.Printf("Connected to %s", pi.ID.String())
+ }
+}
+
+func getPrivKey(nodeId string) (crypto.PrivKey, error) {
+ seed := sha256.Sum256([]byte(nodeId))
+ priv, _, err := crypto.GenerateEd25519Key(bytes.NewReader(seed[:]))
+ if err != nil {
+ return nil, err
+ }
+ return priv, nil
+}
+
+func getNode(ctx context.Context) {
+ once.Do(func() {
+ nodeId := GetNodeId()
+ var opts []libp2p.Option
+ priv, err := getPrivKey(nodeId)
+ if err != nil {
+ log.Fatalf("failed to generate key: %v", err)
+ }
+ opts = append(opts, libp2p.Identity(priv))
+ opts = append(opts, libp2p.Security(noise.ID, noise.New))
+
+ pskHash := sha256.Sum256([]byte("forwarder_network"))
+ psk := pnet.PSK(pskHash[:])
+ opts = append(opts, libp2p.PrivateNetwork(psk))
+
+ // Performance optimizations
+ opts = append(opts, libp2p.ConnectionManager(nil)) // No connection limits
+ opts = append(opts, libp2p.EnableHolePunching()) // Better NAT traversal
+ opts = append(opts, libp2p.EnableRelay()) // Allow relaying
+
+ node, err = libp2p.New(opts...)
+ if err != nil {
+ log.Fatalf("failed to create host: %v", err)
+ }
+
+ // Configure GossipSub for better performance
+ gossipOpts := []pubsub.Option{
+ pubsub.WithMessageSigning(false), // Disable message signing for speed
+ pubsub.WithStrictSignatureVerification(false), // Disable signature verification
+ pubsub.WithMaxMessageSize(1024 * 1024), // 1MB max message size for batches
+ pubsub.WithValidateQueueSize(1000), // Larger validation queue
+ pubsub.WithPeerOutboundQueueSize(1000), // Larger peer queues
+ }
+
+ ps, err = pubsub.NewGossipSub(ctx, node, gossipOpts...)
+ if err != nil {
+ node.Close()
+ log.Fatalf("failed to create pubsub: %v", err)
+ }
+
+ rendezvous := "forwarder_network"
+ notifee := &discoveryNotifee{h: node}
+ mdnsSer = mdns.NewMdnsService(node, rendezvous, notifee)
+ if err := mdnsSer.Start(); err != nil {
+ node.Close()
+ log.Fatalf("failed to start mdns service: %v", err)
+ }
+ })
+}
+
+type libP2PConnector struct {
+ topic string
+ sub *pubsub.Subscription
+ subResend *pubsub.Subscription
+ top *pubsub.Topic
+ topResend *pubsub.Topic
+ ctx context.Context
+ cancel context.CancelFunc
+
+ // Async publishing
+ writeChan chan RecordData
+ batchSize int
+ batchTimeout time.Duration
+ workerPool int
+}
+
+func newLibP2PConnector(topic string, ctx context.Context, cancel context.CancelFunc) *libP2PConnector {
+ getNode(ctx)
+ mu.Lock()
+ var err error
+ t, ok := topicsMap[topic]
+ if !ok {
+ t, err = ps.Join(topic)
+ if err != nil {
+ mu.Unlock()
+ log.Fatalf("failed to join topic %s: %v", topic, err)
+ }
+ topicsMap[topic] = t
+ }
+
+ t2, okResend := topicsMap[topic+"/resend"]
+ if !okResend {
+ t2, err = ps.Join(topic + "/resend")
+ if err != nil {
+ mu.Unlock()
+ log.Fatalf("failed to join topic %s: %v", topic+"/resend", err)
+ }
+ topicsMap[topic+"/resend"] = t2
+ }
+
+ refCount++
+ mu.Unlock()
+
+ connector := &libP2PConnector{
+ topic: topic,
+ top: t,
+ topResend: t2,
+ ctx: ctx,
+ cancel: cancel,
+ writeChan: make(chan RecordData, 2000),
+ batchSize: 100,
+ batchTimeout: 10 * time.Millisecond,
+ workerPool: 5,
+ }
+
+ connector.startAsyncPublishers()
+
+ return connector
+}
+
+func (c *libP2PConnector) tail(handler func(record RecordData) error) {
+ sub, err := c.top.Subscribe()
+ if err != nil {
+ log.Fatalf("failed to subscribe to topic %s: %v", c.topic, err)
+ }
+ c.sub = sub
+ go handleRecordSub(c.sub, c.ctx, handler)
+}
+
+func (c *libP2PConnector) tailResend(handler func(data ResendRequest) error) {
+ sub, err := c.topResend.Subscribe()
+ if err != nil {
+ log.Fatalf("failed to subscribe to topic %s: %v", c.topic, err)
+ }
+ c.subResend = sub
+ go handleSub(c.subResend, c.ctx, handler)
+}
+
+func handleSub[T any](sub *pubsub.Subscription, ctx context.Context, handler func(data T) error) {
+ for {
+ msg, err := sub.Next(ctx)
+ if err != nil {
+ if err == context.Canceled {
+ return
+ }
+ log.Printf("subscription error for topic %s: %v", sub.Topic(), err)
+ return
+ }
+ var rec T
+ err = json.Unmarshal(msg.Data, &rec)
+ if err != nil {
+ log.Printf("unmarshal error for topic %s: %v", sub.Topic(), err)
+ continue
+ }
+ if handler != nil {
+ if err := handler(rec); err != nil {
+ log.Printf("handler error for topic %s: %v", sub.Topic(), err)
+ }
+ }
+ }
+}
+
+func handleRecordSub(sub *pubsub.Subscription, ctx context.Context, handler func(record RecordData) error) {
+ for {
+ msg, err := sub.Next(ctx)
+ if err != nil {
+ if err == context.Canceled {
+ return
+ }
+ log.Printf("subscription error for topic %s: %v", sub.Topic(), err)
+ return
+ }
+
+ // Try to unmarshal as batch first
+ var batch BatchRecord
+ if err := json.Unmarshal(msg.Data, &batch); err == nil && len(batch.Records) > 0 {
+ // Handle batched records
+ for _, record := range batch.Records {
+ if handler != nil {
+ if err := handler(record); err != nil {
+ log.Printf("handler error for batched record: %v", err)
+ }
+ }
+ }
+ continue
+ }
+
+ // Try to unmarshal as single record (backwards compatibility)
+ var record RecordData
+ if err := json.Unmarshal(msg.Data, &record); err == nil {
+ if handler != nil {
+ if err := handler(record); err != nil {
+ log.Printf("handler error for single record: %v", err)
+ }
+ }
+ continue
+ }
+
+ log.Printf("failed to unmarshal message as batch or single record for topic %s", sub.Topic())
+ }
+}
+
+func (c *libP2PConnector) startAsyncPublishers() {
+ // Start worker pool for batched async publishing
+ for i := 0; i < c.workerPool; i++ {
+ go c.publishWorker()
+ }
+}
+
+func (c *libP2PConnector) publishWorker() {
+ batch := make([]RecordData, 0, c.batchSize)
+ timer := time.NewTimer(c.batchTimeout)
+ timer.Stop()
+
+ for {
+ select {
+ case <-c.ctx.Done():
+ // Flush final batch
+ if len(batch) > 0 {
+ err := c.publishBatch(batch)
+ if err != nil {
+ log.Printf("Error publishing batch: %v", err)
+ }
+ }
+ return
+
+ case record := <-c.writeChan:
+ batch = append(batch, record)
+
+ // Check if we should flush
+ if len(batch) >= c.batchSize {
+ err := c.publishBatch(batch)
+ if err != nil {
+ log.Printf("Error publishing batch: %v", err)
+ }
+ batch = batch[:0]
+ timer.Stop()
+ } else if len(batch) == 1 {
+ // First record in batch, start timer
+ timer.Reset(c.batchTimeout)
+ }
+
+ case <-timer.C:
+ // Timer expired, flush whatever we have
+ if len(batch) > 0 {
+ err := c.publishBatch(batch)
+ if err != nil {
+ log.Printf("Error publishing batch: %v", err)
+ }
+ batch = batch[:0]
+ }
+ }
+ }
+}
+
+func (c *libP2PConnector) publishBatch(records []RecordData) error {
+ if len(records) == 0 {
+ return nil
+ }
+
+ // Create batch record
+ batchRecord := BatchRecord{Records: records}
+
+ data, err := json.Marshal(batchRecord)
+ if err != nil {
+ return err
+ }
+
+ // Publish with timeout to prevent blocking
+ go func() {
+ pubCtx, pubCancel := context.WithTimeout(c.ctx, 100*time.Millisecond)
+ defer pubCancel()
+
+ if err := c.top.Publish(pubCtx, data); err != nil {
+ if err != context.DeadlineExceeded {
+ log.Printf("Error publishing batch of %d records: %v", len(records), err)
+ }
+ }
+ }()
+ return nil
+}
+
+func (c *libP2PConnector) write(record RecordData) error {
+ select {
+ case c.writeChan <- record:
+ return nil
+ case <-c.ctx.Done():
+ return c.ctx.Err()
+ default:
+ // Channel full, try to publish directly
+ return c.publishSingle(record)
+ }
+}
+
+func (c *libP2PConnector) publishSingle(record RecordData) error {
+ if c.top == nil {
+ return context.Canceled
+ }
+ data, err := json.Marshal(record)
+ if err != nil {
+ return err
+ }
+ return c.top.Publish(c.ctx, data)
+}
+
+func (c *libP2PConnector) writeResend(req ResendRequest) error {
+ if c.topResend == nil {
+ return context.Canceled
+ }
+ data, err := json.Marshal(req)
+ if err != nil {
+ return err
+ }
+ return c.topResend.Publish(c.ctx, data)
+}
+
+func (c *libP2PConnector) close() error {
+ mu.Lock()
+ refCount--
+ closeHost := refCount == 0
+ mu.Unlock()
+
+ if c.cancel != nil {
+ c.cancel()
+ }
+ if c.sub != nil {
+ c.sub.Cancel()
+ }
+ if c.subResend != nil {
+ c.subResend.Cancel()
+ }
+ if closeHost {
+ // close all topics when shutting down host
+ for _, top := range topicsMap {
+ _ = top.Close()
+ }
+ topicsMap = make(map[string]*pubsub.Topic)
+ }
+
+ c.top = nil
+
+ if !closeHost {
+ return nil
+ }
+
+ if mdnsSer != nil {
+ _ = mdnsSer.Close()
+ mdnsSer = nil
+ }
+
+ var err error
+ if node != nil {
+ err = node.Close()
+ }
+
+ node = nil
+ ps = nil
+ refCount = 0
+ once = sync.Once{}
+
+ return err
+}
+
+func (c *libP2PConnector) getType() string {
+ return "libp2p"
+}
diff --git a/networking/forwarder/src/libp2p_test.go b/networking/forwarder/src/libp2p_test.go
new file mode 100644
index 00000000..3cbbb3fc
--- /dev/null
+++ b/networking/forwarder/src/libp2p_test.go
@@ -0,0 +1,175 @@
+package forwarder
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestLibP2PConnectorCreation(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ conn := newLibP2PConnector("test_topic", ctx, cancel)
+ assert.NotNil(t, conn)
+ assert.Equal(t, "test_topic", conn.topic)
+ assert.NotNil(t, conn.top)
+ assert.Nil(t, conn.sub)
+ err := conn.close()
+ assert.NoError(t, err)
+}
+
+func TestLibP2PConnectorGetType(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ conn := newLibP2PConnector("test_topic", ctx, cancel)
+ assert.Equal(t, "libp2p", conn.getType())
+ err := conn.close()
+ assert.NoError(t, err)
+}
+
+func TestLibP2PConnectorTailAndWriteSameTopic(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ conn := newLibP2PConnector("test_topic_tail_and_write", ctx, cancel)
+
+ received := make(chan RecordData, 1)
+ errChan := make(chan error, 1)
+
+ conn.tail(func(rec RecordData) error {
+ received <- rec
+ return nil
+ })
+
+ time.Sleep(100 * time.Millisecond)
+
+ rec := RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{
+ SourceNodeId: "test_node_id",
+ SourcePath: "test_path",
+ },
+ SourceRowID: 1,
+ SourceTimestamp: time.Now(),
+ },
+ Data: map[string]interface{}{"test_key": "test_value"},
+ }
+ err := conn.write(rec)
+ require.NoError(t, err)
+
+ select {
+ case got := <-received:
+ assert.Equal(t, rec.SourceKey.SourceNodeId, got.SourceKey.SourceNodeId)
+ assert.Equal(t, rec.SourceKey.SourcePath, got.SourceKey.SourcePath)
+ assert.Equal(t, rec.SourceRowID, got.SourceRowID)
+ assert.Equal(t, rec.Data, got.Data)
+ assert.WithinDuration(t, rec.SourceTimestamp, got.SourceTimestamp, time.Second)
+ case err := <-errChan:
+ t.Fatalf("handler error: %v", err)
+ case <-time.After(2 * time.Second):
+ t.Fatal("timeout waiting for message")
+ }
+
+ err = conn.close()
+ assert.NoError(t, err)
+}
+
+func TestLibP2PConnectorTailAndWriteDifferentTopic(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ conn1 := newLibP2PConnector("test_topic_tail_and_write1", ctx, cancel)
+ conn2 := newLibP2PConnector("test_topic_tail_and_write2", ctx, cancel)
+
+ received := make(chan RecordData, 1)
+
+ conn1.tail(func(rec RecordData) error {
+ received <- rec
+ return nil
+ })
+
+ time.Sleep(100 * time.Millisecond)
+
+ rec := RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{
+ SourceNodeId: "test_node_id",
+ SourcePath: "test_path",
+ },
+ SourceRowID: 1,
+ SourceTimestamp: time.Now(),
+ },
+ Data: map[string]interface{}{"test_key": "test_value"},
+ }
+ err := conn2.write(rec)
+ require.NoError(t, err)
+
+ select {
+ case <-received:
+ t.Fatal("should not receive message from different topic")
+ case <-time.After(500 * time.Millisecond):
+ }
+
+ err = conn1.close()
+ assert.NoError(t, err)
+ err = conn2.close()
+ assert.NoError(t, err)
+}
+
+func TestLibP2PConnectorMultipleSubscriptionsSameTopic(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ conn1 := newLibP2PConnector("test_topic_multiple_subscriptions", ctx, cancel)
+ conn2 := newLibP2PConnector("test_topic_multiple_subscriptions", ctx, cancel)
+
+ received1 := make(chan RecordData, 1)
+ received2 := make(chan RecordData, 1)
+
+ conn1.tail(func(rec RecordData) error {
+ received1 <- rec
+ return nil
+ })
+ conn2.tail(func(rec RecordData) error {
+ received2 <- rec
+ return nil
+ })
+
+ time.Sleep(100 * time.Millisecond)
+
+ rec := RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{
+ SourceNodeId: "test_node_id",
+ SourcePath: "test_path",
+ },
+ SourceRowID: 1,
+ SourceTimestamp: time.Now(),
+ },
+ Data: map[string]interface{}{"test_key": "test_value"},
+ }
+ err := conn1.write(rec)
+ require.NoError(t, err)
+
+ select {
+ case got := <-received1:
+ assert.Equal(t, rec.SourceKey.SourceNodeId, got.SourceKey.SourceNodeId)
+ assert.Equal(t, rec.SourceKey.SourcePath, got.SourceKey.SourcePath)
+ assert.Equal(t, rec.SourceRowID, got.SourceRowID)
+ assert.Equal(t, rec.Data, got.Data)
+ assert.WithinDuration(t, rec.SourceTimestamp, got.SourceTimestamp, time.Second)
+ case <-time.After(2 * time.Second):
+ t.Fatal("timeout waiting for message on conn1")
+ }
+
+ select {
+ case got := <-received2:
+ assert.Equal(t, rec.SourceKey.SourceNodeId, got.SourceKey.SourceNodeId)
+ assert.Equal(t, rec.SourceKey.SourcePath, got.SourceKey.SourcePath)
+ assert.Equal(t, rec.SourceRowID, got.SourceRowID)
+ assert.Equal(t, rec.Data, got.Data)
+ assert.WithinDuration(t, rec.SourceTimestamp, got.SourceTimestamp, time.Second)
+ case <-time.After(2 * time.Second):
+ t.Fatal("timeout waiting for message on conn2")
+ }
+
+ err = conn1.close()
+ assert.NoError(t, err)
+ err = conn2.close()
+ assert.NoError(t, err)
+}
diff --git a/networking/forwarder/src/schema.go b/networking/forwarder/src/schema.go
new file mode 100644
index 00000000..5022468d
--- /dev/null
+++ b/networking/forwarder/src/schema.go
@@ -0,0 +1,72 @@
+package forwarder
+
+import (
+ "context"
+ "time"
+)
+
+type SourceKey struct {
+ SourceNodeId string `json:"source_node_id"`
+ SourcePath string `json:"source_path"` // db:table
+}
+
+type TrackingData struct {
+ SourceKey
+ SourceRowID int64 `json:"source_row_id"`
+ SourceTimestamp time.Time `json:"source_timestamp"`
+}
+type RecordData struct {
+ TrackingData
+ Data map[string]interface{} `json:"data"`
+}
+
+type BatchRecord struct {
+ Records []RecordData `json:"records"`
+}
+
+type ForwardingPair struct {
+ source connection
+ sink connection
+}
+
+type connection interface {
+ tail(handler func(record RecordData) error)
+ write(record RecordData) error
+ close() error
+ getType() string
+}
+
+type LibP2PConnection interface {
+ connection
+ tailResend(handler func(record ResendRequest) error)
+ writeResend(record ResendRequest) error
+}
+
+type SQLiteConnection interface {
+ connection
+ getLatestRowIds() (map[SourceKey]int64, error)
+ readRange(start, end int64) ([]RecordData, error)
+ getNodeId() string
+ getTablePath() string
+}
+
+type GapRange struct {
+ Start int64 `json:"start"`
+ End int64 `json:"end"`
+}
+type ResendRequest struct {
+ SourceNodeID string `json:"source_node_id"`
+ SourcePath string `json:"source_path"`
+ Gaps []GapRange `json:"gaps"`
+}
+
+type stateStoreInterface interface {
+ onRecord(record RecordData)
+ getWriteableMessages() []RecordData
+ getResendRequests() []ResendRequest
+ getCurrentGaps() map[SourceKey][]gap
+}
+
+type Forwarder interface {
+ Start(ctx context.Context) error
+}
diff --git a/networking/forwarder/src/sqlite.go b/networking/forwarder/src/sqlite.go
new file mode 100644
index 00000000..7a449f61
--- /dev/null
+++ b/networking/forwarder/src/sqlite.go
@@ -0,0 +1,649 @@
+package forwarder
+
+import (
+ "database/sql"
+ "errors"
+ "fmt"
+ "log"
+ "reflect"
+ "sort"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+
+ _ "github.com/mattn/go-sqlite3"
+)
+
+type sqliteConnector struct {
+ db *sql.DB
+ tableName string
+ stop chan struct{}
+ wg sync.WaitGroup
+ pendingWrites []RecordData
+ mu sync.Mutex
+ nodeId string
+ tablePath string
+ // Cache the original columns (non-tracking columns)
+ originalColumns []string
+ columnTypes map[string]string
+}
+
+func newSQLiteConnector(dbPath, tableName string) (*sqliteConnector, error) {
+ if tableName == "" {
+ return nil, errors.New("table name cannot be empty")
+ }
+ db, err := sql.Open("sqlite3", dbPath)
+ if err != nil {
+ return nil, err
+ }
+ _, err = db.Exec("PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL; PRAGMA busy_timeout = 500; PRAGMA cache_size = -64000;")
+ if err != nil {
+ db.Close()
+ return nil, fmt.Errorf("failed to apply PRAGMA settings: %w", err)
+ }
+
+ // Increase connection pool for better concurrency
+ db.SetMaxOpenConns(25)
+ db.SetMaxIdleConns(10)
+ db.SetConnMaxLifetime(5 * time.Minute)
+
+ c := &sqliteConnector{
+ db: db,
+ tableName: tableName,
+ stop: make(chan struct{}),
+ pendingWrites: []RecordData{},
+ nodeId: GetNodeId(),
+ tablePath: dbPath + ":" + tableName,
+ columnTypes: make(map[string]string),
+ }
+
+ // Get the table schema before adding tracking columns
+ err = c.loadTableSchema()
+ if err != nil && !strings.Contains(err.Error(), "no such table") {
+ db.Close()
+ return nil, err
+ }
+
+ err = c.ensureTrackingColumns()
+ if err != nil {
+ db.Close()
+ return nil, err
+ }
+
+ // Reload schema after ensuring tracking columns
+ err = c.loadTableSchema()
+ if err != nil {
+ db.Close()
+ return nil, err
+ }
+
+ c.wg.Add(1)
+ go func() {
+ defer c.wg.Done()
+ c.writerLoop()
+ }()
+ return c, nil
+}
+
+func (c *sqliteConnector) loadTableSchema() error {
+ rows, err := c.db.Query(fmt.Sprintf(`PRAGMA table_info("%s")`, c.tableName))
+ if err != nil {
+ return err
+ }
+ defer rows.Close()
+
+ trackingCols := make(map[string]bool)
+ for _, col := range []string{"source_node_id", "source_path", "source_row_id", "source_timestamp"} {
+ trackingCols[col] = true
+ }
+
+ c.originalColumns = []string{}
+ c.columnTypes = make(map[string]string)
+
+ for rows.Next() {
+ var cid int
+ var name string
+ var typ string
+ var notnull int
+ var dflt interface{}
+ var pk int
+ if err := rows.Scan(&cid, &name, &typ, ¬null, &dflt, &pk); err != nil {
+ return err
+ }
+
+ c.columnTypes[name] = typ
+
+ // Only include non-tracking columns in originalColumns
+ if !trackingCols[name] {
+ c.originalColumns = append(c.originalColumns, name)
+ }
+ }
+
+ return nil
+}
+
+func (c *sqliteConnector) getNodeId() string {
+ return c.nodeId
+}
+
+func (c *sqliteConnector) getTablePath() string {
+ return c.tablePath
+}
+
+func (c *sqliteConnector) writerLoop() {
+ ticker := time.NewTicker(50 * time.Millisecond)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ c.mu.Lock()
+ batch := c.pendingWrites
+ c.pendingWrites = nil
+ c.mu.Unlock()
+ if len(batch) > 0 {
+ if err := c.writeBatch(batch); err != nil {
+ log.Printf("Error writing batch: %v", err)
+ }
+ }
+ case <-c.stop:
+ c.mu.Lock()
+ batch := c.pendingWrites
+ c.pendingWrites = nil
+ c.mu.Unlock()
+ if len(batch) > 0 {
+ if err := c.writeBatch(batch); err != nil {
+ log.Printf("Error writing final batch: %v", err)
+ }
+ }
+ return
+ }
+ }
+}
+
+func (c *sqliteConnector) writeBatch(records []RecordData) error {
+ if len(records) == 0 {
+ return nil
+ }
+ tx, err := c.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+
+ // Build column list: tracking columns + original columns
+ trackingCols := []string{"source_node_id", "source_path", "source_row_id", "source_timestamp"}
+ cols := append(trackingCols, c.originalColumns...)
+ colStr := strings.Join(cols, `", "`)
+
+ places := make([]string, len(cols))
+ for i := range places {
+ places[i] = "?"
+ }
+ singlePlace := "(" + strings.Join(places, ", ") + ")"
+ rowPlaces := make([]string, len(records))
+ for i := range rowPlaces {
+ rowPlaces[i] = singlePlace
+ }
+ valuesStr := strings.Join(rowPlaces, ", ")
+
+ query := fmt.Sprintf(`INSERT INTO "%s" ("%s") VALUES %s`, c.tableName, colStr, valuesStr)
+
+ vals := make([]interface{}, 0, len(records)*len(cols))
+ for _, rec := range records {
+ // Add tracking columns
+ vals = append(vals, rec.SourceNodeId, rec.SourcePath, rec.SourceRowID, rec.SourceTimestamp)
+
+ // Add original column values from Data map
+ for _, col := range c.originalColumns {
+ if val, ok := rec.Data[col]; ok {
+ vals = append(vals, val)
+ } else {
+ vals = append(vals, nil)
+ }
+ }
+ }
+
+ _, err = tx.Exec(query, vals...)
+ if err != nil {
+ return err
+ }
+ return tx.Commit()
+}
+
+func (c *sqliteConnector) ensureTrackingColumns() error {
+ // Wrap table creation and alterations in a transaction for atomicity
+ tx, err := c.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+
+ // Check if table exists
+ var count int
+ err = tx.QueryRow(`SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = ?`, c.tableName).Scan(&count)
+ if err != nil {
+ return err
+ }
+ if count == 0 {
+ // Create table with only tracking columns initially
+ // The original schema should be defined by the first records written
+ typePairs := getJsonTagsWithSqliteTypes(reflect.TypeOf(TrackingData{}))
+ colDefs := make([]string, 0, len(typePairs))
+ for _, pair := range typePairs {
+ colDefs = append(colDefs, fmt.Sprintf("%s %s", pair.name, pair.typeStr))
+ }
+ createQuery := fmt.Sprintf(`CREATE TABLE "%s" (%s)`, c.tableName, strings.Join(colDefs, ", "))
+ _, err := tx.Exec(createQuery)
+ if err != nil {
+ return err
+ }
+ } else {
+ // Table exists, ensure tracking columns
+ existing := make(map[string]bool)
+ rows, err := tx.Query(fmt.Sprintf(`PRAGMA table_info("%s")`, c.tableName))
+ if err != nil {
+ return err
+ }
+ defer rows.Close()
+ for rows.Next() {
+ var cid int
+ var name string
+ var typ string
+ var notnull int
+ var dflt interface{}
+ var pk int
+ if err := rows.Scan(&cid, &name, &typ, ¬null, &dflt, &pk); err != nil {
+ return err
+ }
+ existing[name] = true
+ }
+
+ typePairs := getJsonTagsWithSqliteTypes(reflect.TypeOf(TrackingData{}))
+ for _, pair := range typePairs {
+ if !existing[pair.name] {
+ if _, err := tx.Exec(fmt.Sprintf(`ALTER TABLE "%s" ADD COLUMN %s %s`, c.tableName, pair.name, pair.typeStr)); err != nil {
+ return err
+ }
+ }
+ }
+ }
+
+ return tx.Commit()
+}
+
+func (c *sqliteConnector) getLatestRowIds() (map[SourceKey]int64, error) {
+ keyCols := getJsonTagNames(reflect.TypeOf(SourceKey{}))
+ rowIdField := "SourceRowID"
+ rowIDCol := getFieldJsonTag(reflect.TypeOf(TrackingData{}), rowIdField)
+ if rowIDCol == "" {
+ return nil, fmt.Errorf("could not find field %s in TrackingData struct", rowIdField)
+ }
+
+ selectCols := strings.Join(keyCols, ", ")
+ query := fmt.Sprintf(`SELECT %s, MAX(%s) FROM "%s" GROUP BY %s`, selectCols, rowIDCol, c.tableName, selectCols)
+
+ rows, err := c.db.Query(query)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ m := make(map[SourceKey]int64)
+ for rows.Next() {
+ strPtrs := make([]*string, len(keyCols))
+ scanArgs := make([]interface{}, 0, len(keyCols)+1)
+ for i := range keyCols {
+ var s string
+ strPtrs[i] = &s
+ scanArgs = append(scanArgs, &s)
+ }
+ var maxPtr int64
+ scanArgs = append(scanArgs, &maxPtr)
+ if err := rows.Scan(scanArgs...); err != nil {
+ return nil, err
+ }
+ var key SourceKey
+ val := reflect.ValueOf(&key).Elem()
+ keyType := reflect.TypeOf(key)
+ for i, colName := range keyCols {
+ // find field with json tag = colName
+ for f := 0; f < keyType.NumField(); f++ {
+ field := keyType.Field(f)
+ tag := strings.Split(field.Tag.Get("json"), ",")[0]
+ if tag == "" {
+ tag = strings.ToLower(field.Name)
+ }
+ if tag == colName {
+ if strPtrs[i] != nil {
+ val.FieldByName(field.Name).SetString(*strPtrs[i])
+ }
+ break
+ }
+ }
+ }
+ m[key] = maxPtr
+ }
+ return m, nil
+}
+
+func (c *sqliteConnector) scanToRecord(rows *sql.Rows) (RecordData, int64, error) {
+ // Get column names from the result set
+ columns, err := rows.Columns()
+ if err != nil {
+ return RecordData{}, 0, err
+ }
+
+ // Create scan destinations
+ scanArgs := make([]interface{}, len(columns))
+ values := make([]interface{}, len(columns))
+ for i := range values {
+ scanArgs[i] = &values[i]
+ }
+
+ err = rows.Scan(scanArgs...)
+ if err != nil {
+ return RecordData{}, 0, err
+ }
+
+ var rec RecordData
+ rec.Data = make(map[string]interface{})
+ var rowID int64
+
+ // Process each column
+ for i, col := range columns {
+ val := values[i]
+
+ // Handle NULL values
+ if val == nil {
+ continue
+ }
+
+ // Convert []byte to appropriate type
+ if b, ok := val.([]byte); ok {
+ val = string(b)
+ }
+
+ switch col {
+ case "source_node_id":
+ if s, ok := val.(string); ok {
+ rec.SourceNodeId = s
+ }
+ case "source_path":
+ if s, ok := val.(string); ok {
+ rec.SourcePath = s
+ }
+ case "source_row_id":
+ switch v := val.(type) {
+ case int64:
+ rec.SourceRowID = v
+ case int:
+ rec.SourceRowID = int64(v)
+ case string:
+ if parsed, err := strconv.ParseInt(v, 10, 64); err == nil {
+ rec.SourceRowID = parsed
+ }
+ }
+ case "source_timestamp":
+ switch v := val.(type) {
+ case time.Time:
+ rec.SourceTimestamp = v
+ case string:
+ if parsed, err := time.Parse(time.RFC3339Nano, v); err == nil {
+ rec.SourceTimestamp = parsed
+ } else if parsed, err := time.Parse("2006-01-02 15:04:05", v); err == nil {
+ rec.SourceTimestamp = parsed
+ }
+ }
+ case "rowid":
+ switch v := val.(type) {
+ case int64:
+ rowID = v
+ case int:
+ rowID = int64(v)
+ }
+ default:
+ // All other columns go into the Data map
+ rec.Data[col] = val
+ }
+ }
+
+ return rec, rowID, nil
+}
+
+func (c *sqliteConnector) readRange(start, end int64) ([]RecordData, error) {
+ // Select all columns plus rowid
+ query := fmt.Sprintf(`SELECT *, rowid FROM "%s" WHERE rowid >= ? AND rowid <= ? ORDER BY rowid`, c.tableName)
+ rows, err := c.db.Query(query, start, end)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ var records []RecordData
+ for rows.Next() {
+ rec, rowID, err := c.scanToRecord(rows)
+ if err != nil {
+ return nil, err
+ }
+ // Override tracking data so that this table is treated as the new source
+ rec.SourceNodeId = c.nodeId
+ rec.SourcePath = c.tablePath
+ rec.SourceRowID = rowID
+ rec.SourceTimestamp = time.Now()
+ records = append(records, rec)
+ }
+ return records, nil
+}
+
+func (c *sqliteConnector) tail(handler func(record RecordData) error) {
+ c.wg.Add(1)
+ go func() {
+ defer c.wg.Done()
+ var last int64
+ err := c.db.QueryRow(fmt.Sprintf(`SELECT IFNULL(MAX(rowid), 0) FROM "%s"`, c.tableName)).Scan(&last)
+ if err != nil {
+ last = 0
+ }
+ // Prepare the statement outside the loop for efficiency
+ query := fmt.Sprintf(`SELECT *, rowid FROM "%s" WHERE rowid > ? ORDER BY rowid LIMIT ?`, c.tableName)
+ stmt, err := c.db.Prepare(query)
+ if err != nil {
+ log.Printf("Error preparing tail statement: %v", err)
+ return
+ }
+ defer stmt.Close()
+
+ // Adaptive polling: start fast, slow down when idle
+ minPollInterval := 1 * time.Millisecond
+ maxPollInterval := 50 * time.Millisecond
+ currentInterval := minPollInterval
+ batchSize := 500 // Process records in larger batches for better throughput
+
+ for {
+ select {
+ case <-c.stop:
+ return
+ default:
+ }
+ rows, err := stmt.Query(last, batchSize)
+ if err != nil {
+ time.Sleep(currentInterval)
+ continue
+ }
+ hadNew := false
+ recordCount := 0
+ for rows.Next() {
+ rec, rowID, err := c.scanToRecord(rows)
+ if err != nil {
+ log.Printf("Error scanning record: %v", err)
+ break
+ }
+ // Override tracking data so that this table is treated as the new source
+ rec.SourceNodeId = c.nodeId
+ rec.SourcePath = c.tablePath
+ rec.SourceRowID = rowID
+ rec.SourceTimestamp = time.Now()
+ last = rowID
+ err = handler(rec)
+ if err != nil {
+ log.Printf("Error handling record: %v", err)
+ }
+ hadNew = true
+ recordCount++
+ }
+ rows.Close()
+
+ // Adaptive interval adjustment
+ if hadNew {
+ // Had records, speed up polling
+ currentInterval = minPollInterval
+ if recordCount == batchSize {
+ // Full batch, poll immediately
+ continue
+ }
+ } else {
+ // No records, slow down gradually
+ currentInterval = time.Duration(float64(currentInterval) * 1.5)
+ if currentInterval > maxPollInterval {
+ currentInterval = maxPollInterval
+ }
+ }
+ time.Sleep(currentInterval)
+ }
+ }()
+}
+
+func (c *sqliteConnector) write(record RecordData) error {
+ // If we don't know the schema yet, try to infer it from the first record
+ if len(c.originalColumns) == 0 && len(record.Data) > 0 {
+ c.mu.Lock()
+ if len(c.originalColumns) == 0 {
+ // Infer columns from the data
+ for col := range record.Data {
+ c.originalColumns = append(c.originalColumns, col)
+ }
+ // Sort for consistency
+ sort.Strings(c.originalColumns)
+
+ // Add columns to table if they don't exist
+ tx, err := c.db.Begin()
+ if err == nil {
+ defer tx.Rollback()
+ for col := range record.Data {
+ // Infer SQL type from Go type
+ sqlType := "TEXT" // default
+ switch record.Data[col].(type) {
+ case int, int32, int64:
+ sqlType = "INTEGER"
+ case float32, float64:
+ sqlType = "REAL"
+ case bool:
+ sqlType = "INTEGER"
+ }
+
+ // Try to add column (will fail silently if it exists)
+ tx.Exec(fmt.Sprintf(`ALTER TABLE "%s" ADD COLUMN "%s" %s`, c.tableName, col, sqlType))
+ }
+ tx.Commit()
+ }
+ }
+ c.mu.Unlock()
+ }
+
+ c.mu.Lock()
+ c.pendingWrites = append(c.pendingWrites, record)
+ c.mu.Unlock()
+ return nil
+}
+
+func (c *sqliteConnector) close() error {
+ close(c.stop)
+ c.wg.Wait()
+ return c.db.Close()
+}
+
+func (c *sqliteConnector) getType() string {
+ return "sqlite"
+}
+
+type typedPair struct {
+ name string
+ typeStr string
+}
+
+func getJsonTagsWithSqliteTypes(t reflect.Type) []typedPair {
+ typePairs := []typedPair{}
+ for i := 0; i < t.NumField(); i++ {
+ f := t.Field(i)
+ if f.Anonymous {
+ typePairs = append(typePairs, getJsonTagsWithSqliteTypes(f.Type)...)
+ continue
+ }
+ tag := f.Tag.Get("json")
+ if tag == "-" {
+ continue
+ }
+ if tag != "" {
+ tag = strings.Split(tag, ",")[0]
+ }
+ if tag == "" {
+ tag = strings.ToLower(f.Name)
+ }
+ var sqlType string
+ switch f.Type.Kind() {
+ case reflect.String:
+ sqlType = "TEXT"
+ case reflect.Int, reflect.Int32, reflect.Int64:
+ sqlType = "INTEGER"
+ default:
+ if f.Type == reflect.TypeOf(time.Time{}) {
+ sqlType = "DATETIME"
+ } else {
+ sqlType = "BLOB"
+ }
+ }
+ typePairs = append(typePairs, typedPair{tag, sqlType})
+ }
+ return typePairs
+}
+
+func getJsonTagNames(t reflect.Type) []string {
+ cols := []string{}
+ for i := 0; i < t.NumField(); i++ {
+ f := t.Field(i)
+ if f.Anonymous {
+ cols = append(cols, getJsonTagNames(f.Type)...)
+ continue
+ }
+ tag := strings.Split(f.Tag.Get("json"), ",")[0]
+ if tag == "-" {
+ continue
+ }
+ if tag == "" {
+ tag = strings.ToLower(f.Name)
+ }
+ cols = append(cols, tag)
+ }
+ return cols
+}
+
+func getFieldJsonTag(t reflect.Type, fieldName string) string {
+ for i := 0; i < t.NumField(); i++ {
+ f := t.Field(i)
+ if f.Anonymous {
+ if tag := getFieldJsonTag(f.Type, fieldName); tag != "" {
+ return tag
+ }
+ continue
+ }
+ if f.Name == fieldName {
+ tag := strings.Split(f.Tag.Get("json"), ",")[0]
+ if tag == "" {
+ return strings.ToLower(f.Name)
+ }
+ return tag
+ }
+ }
+ return ""
+}
diff --git a/networking/forwarder/src/sqlite_test.go b/networking/forwarder/src/sqlite_test.go
new file mode 100644
index 00000000..12913948
--- /dev/null
+++ b/networking/forwarder/src/sqlite_test.go
@@ -0,0 +1,236 @@
+package forwarder
+
+import (
+ "fmt"
+ "os"
+ "reflect"
+ "testing"
+ "time"
+
+ "database/sql"
+
+ _ "github.com/mattn/go-sqlite3"
+)
+
+func TestNewSQLiteConnectorCreatesTable(t *testing.T) {
+ c, err := newSQLiteConnector(":memory:", "test_table")
+ if err != nil {
+ t.Fatalf("failed to create connector: %v", err)
+ }
+ defer c.close()
+
+ rows, err := c.db.Query(`PRAGMA table_info("test_table")`)
+ if err != nil {
+ t.Fatalf("failed to query table info: %v", err)
+ }
+ defer rows.Close()
+
+ expectedCols := map[string]string{
+ "source_node_id": "TEXT",
+ "source_path": "TEXT",
+ "source_row_id": "INTEGER",
+ "source_timestamp": "DATETIME",
+ }
+ foundCols := make(map[string]string)
+ for rows.Next() {
+ var cid int
+ var name, typ string
+ var notnull int
+ var dflt interface{}
+ var pk int
+ if err := rows.Scan(&cid, &name, &typ, ¬null, &dflt, &pk); err != nil {
+ t.Fatalf("failed to scan: %v", err)
+ }
+ foundCols[name] = typ
+ }
+ if !reflect.DeepEqual(expectedCols, foundCols) {
+ t.Errorf("expected columns %v, got %v", expectedCols, foundCols)
+ }
+}
+
+func TestEnsureTrackingColumnsAddsMissing(t *testing.T) {
+ db, err := sql.Open("sqlite3", ":memory:")
+ if err != nil {
+ t.Fatalf("failed to open db: %v", err)
+ }
+ _, err = db.Exec(`CREATE TABLE test_table (source_node_id TEXT, data TEXT)`)
+ if err != nil {
+ t.Fatalf("failed to create partial table: %v", err)
+ }
+ db.Close()
+
+ tempDB := t.TempDir() + "/test.db"
+ db, err = sql.Open("sqlite3", tempDB)
+ if err != nil {
+ t.Fatalf("failed to open db: %v", err)
+ }
+ _, err = db.Exec(`CREATE TABLE test_table (source_node_id TEXT, data TEXT)`)
+ if err != nil {
+ t.Fatalf("failed to create partial table: %v", err)
+ }
+ db.Close()
+
+ c, err := newSQLiteConnector(tempDB, "test_table")
+ if err != nil {
+ t.Fatalf("failed to create connector: %v", err)
+ }
+ defer c.close()
+
+ rows, err := c.db.Query(`PRAGMA table_info("test_table")`)
+ if err != nil {
+ t.Fatalf("failed to query table info: %v", err)
+ }
+ defer rows.Close()
+
+ expectedCols := []string{"source_node_id", "data", "source_path", "source_row_id", "source_timestamp"}
+ foundCols := []string{}
+ for rows.Next() {
+ var cid int
+ var name string
+ var typ string
+ var notnull int
+ var dflt interface{}
+ var pk int
+ if err := rows.Scan(&cid, &name, &typ, ¬null, &dflt, &pk); err != nil {
+ t.Fatalf("failed to scan: %v", err)
+ }
+ foundCols = append(foundCols, name)
+ }
+ if len(foundCols) != len(expectedCols) {
+ t.Errorf("expected %d columns, got %d: %v", len(expectedCols), len(foundCols), foundCols)
+ }
+}
+
+func TestWriteAndReadRecord(t *testing.T) {
+ SetNodeId("node1")
+ c, err := newSQLiteConnector("test_write_and_read_db1", "table")
+ if err != nil {
+ t.Fatalf("failed to create connector: %v", err)
+ }
+ defer func() {
+ c.close()
+ os.Remove("test_write_and_read_db1")
+ }()
+
+ rec := RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{
+ SourceNodeId: "node1",
+ SourcePath: "test_write_and_read_db1:table",
+ },
+ SourceRowID: 42,
+ SourceTimestamp: time.Now().UTC(),
+ },
+ Data: map[string]interface{}{
+ "key": "value",
+ "num": 123.45,
+ },
+ }
+ if err := c.write(rec); err != nil {
+ t.Fatalf("failed to write: %v", err)
+ }
+ time.Sleep(200 * time.Millisecond) // Wait for flush
+
+ records, err := c.readRange(1, 999)
+ if err != nil {
+ t.Fatalf("failed to read: %v", err)
+ }
+ if len(records) != 1 {
+ t.Fatalf("expected 1 record, got %d", len(records))
+ }
+ got := records[0]
+ if got.SourceNodeId != rec.SourceNodeId || got.SourcePath != rec.SourcePath || got.SourceRowID != 1 {
+ t.Errorf("tracking data mismatch: got %+v, want %+v", got.TrackingData, rec.TrackingData)
+ }
+ if !reflect.DeepEqual(got.Data, rec.Data) {
+ t.Errorf("data mismatch: got %v, want %v", got.Data, rec.Data)
+ }
+}
+
+func TestTailDetectsWrites(t *testing.T) {
+ SetNodeId("node2")
+ db, errDb := sql.Open("sqlite3", "tail_detects_writes_db2")
+ if errDb != nil {
+ t.Fatalf("failed to open db for alter: %v", errDb)
+ }
+
+ _, errExec := db.Exec("CREATE TABLE table2 (test BOOLEAN)")
+ if errExec != nil {
+ t.Fatalf("failed to create table: %v", errExec)
+ }
+ db.Close()
+
+ c, err := newSQLiteConnector("tail_detects_writes_db2", "table2")
+ if err != nil {
+ t.Fatalf("failed to create connector: %v", err)
+ }
+ defer c.close()
+
+ ch := make(chan RecordData, 1)
+ c.tail(func(r RecordData) error {
+ ch <- r
+ return nil
+ })
+ time.Sleep(100 * time.Millisecond) // Let tail start
+
+ rec := RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node2", SourcePath: "tail_detects_writes_db2:table2"},
+ SourceRowID: 100,
+ SourceTimestamp: time.Now().UTC(),
+ },
+ Data: map[string]interface{}{"test": true},
+ }
+ if err := c.write(rec); err != nil {
+ t.Fatalf("failed to write: %v", err)
+ }
+ time.Sleep(200 * time.Millisecond) // Wait for flush and tail poll
+
+ select {
+ case got := <-ch:
+ if !reflect.DeepEqual(got.Data, rec.Data) {
+ t.Errorf("got %v, want %v", got, rec)
+ }
+ if got.SourceNodeId != rec.SourceNodeId || got.SourcePath != rec.SourcePath || got.SourceRowID != 1 {
+ t.Errorf("tracking data mismatch: got %+v, want %+v", got.TrackingData, rec.TrackingData)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("timeout waiting for tail handler")
+ }
+ os.Remove("tail_detects_writes_db2")
+ os.Remove("tail_detects_writes_db2-wal")
+ os.Remove("tail_detects_writes_db2-shm")
+
+}
+
+func TestBatchWriteMultipleEdge(t *testing.T) {
+ c, err := newSQLiteConnector(":memory:", "test_table")
+ if err != nil {
+ t.Fatalf("failed to create connector: %v", err)
+ }
+ defer c.close()
+
+ for i := 0; i < 3; i++ {
+ rec := RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: fmt.Sprintf("node%d", i), SourcePath: ""},
+ SourceRowID: int64(i),
+ SourceTimestamp: time.Time{},
+ },
+ Data: nil, // Edge: nil Data
+ }
+ if err := c.write(rec); err != nil {
+ t.Fatalf("failed to write: %v", err)
+ }
+ }
+ time.Sleep(200 * time.Millisecond)
+
+ var count int
+ err = c.db.QueryRow(`SELECT COUNT(*) FROM "test_table"`).Scan(&count)
+ if err != nil {
+ t.Fatalf("failed to count: %v", err)
+ }
+ if count != 3 {
+ t.Errorf("expected 3 rows, got %d", count)
+ }
+}
diff --git a/networking/forwarder/src/state_store.go b/networking/forwarder/src/state_store.go
new file mode 100644
index 00000000..f4dc960c
--- /dev/null
+++ b/networking/forwarder/src/state_store.go
@@ -0,0 +1,240 @@
+package forwarder
+
+import (
+ "sort"
+ "sync"
+ "time"
+)
+
+const gracePeriod = 5 * time.Second
+
+type gap struct {
+ GapRange
+ firstSeen time.Time
+ lastRequestSent time.Time
+ timesRequestSent int
+}
+
+type pendingRecordsRange struct {
+ start int64
+ end int64
+ records map[int64]RecordData
+}
+
+func (g gap) isResendable() bool {
+ currentTime := time.Now()
+ if currentTime.Before(g.firstSeen.Add(gracePeriod)) {
+ return false
+ }
+ backoff := gracePeriod * (1 << g.timesRequestSent)
+ return currentTime.After(g.lastRequestSent.Add(backoff))
+}
+
+type stateStore struct {
+ mu sync.RWMutex
+ sourceKeyMu map[SourceKey]*sync.Mutex
+ lastContiguousRowId map[SourceKey]int64
+ recordsToWrite []RecordData
+ gaps map[SourceKey][]gap
+ pending map[SourceKey][]pendingRecordsRange
+}
+
+func newStateStore(lastWrittenRowId map[SourceKey]int64) *stateStore {
+ return &stateStore{
+ lastContiguousRowId: lastWrittenRowId,
+ recordsToWrite: []RecordData{},
+ gaps: make(map[SourceKey][]gap),
+ pending: make(map[SourceKey][]pendingRecordsRange),
+ sourceKeyMu: make(map[SourceKey]*sync.Mutex),
+ }
+}
+
+func (s *stateStore) onRecord(record RecordData) {
+ sk := SourceKey{SourceNodeId: record.SourceNodeId, SourcePath: record.SourcePath}
+
+ s.mu.Lock()
+ if _, ok := s.sourceKeyMu[sk]; !ok {
+ s.sourceKeyMu[sk] = &sync.Mutex{}
+ if _, ok := s.lastContiguousRowId[sk]; !ok {
+ s.lastContiguousRowId[sk] = 0
+ }
+ s.gaps[sk] = []gap{}
+ s.pending[sk] = []pendingRecordsRange{}
+ }
+ s.mu.Unlock()
+ s.sourceKeyMu[sk].Lock()
+ defer s.sourceKeyMu[sk].Unlock()
+ l := s.lastContiguousRowId[sk]
+ r := record.SourceRowID
+ if r <= l {
+ return
+ }
+
+ for _, ru := range s.pending[sk] {
+ if _, has := ru.records[r]; has {
+ return
+ }
+ }
+
+ currentHighest := l
+ for _, ru := range s.pending[sk] {
+ if ru.end > currentHighest {
+ currentHighest = ru.end
+ }
+ }
+
+ gaps := s.gaps[sk]
+ newGaps := []gap{}
+ filled := false
+ for _, g := range gaps {
+ if g.Start <= r && r <= g.End {
+ filled = true
+ if g.Start < r {
+ newGaps = append(newGaps, gap{GapRange: GapRange{Start: g.Start, End: r - 1}, firstSeen: g.firstSeen, lastRequestSent: g.lastRequestSent, timesRequestSent: g.timesRequestSent})
+ }
+ if r < g.End {
+ newGaps = append(newGaps, gap{GapRange: GapRange{Start: r + 1, End: g.End}, firstSeen: g.firstSeen, lastRequestSent: g.lastRequestSent, timesRequestSent: g.timesRequestSent})
+ }
+ } else {
+ newGaps = append(newGaps, g)
+ }
+ }
+ s.gaps[sk] = mergeGaps(newGaps)
+
+ if !filled && r > currentHighest+1 {
+ gr := GapRange{Start: currentHighest + 1, End: r - 1}
+ if gr.Start <= gr.End {
+ newG := gap{GapRange: gr, firstSeen: time.Now(), lastRequestSent: time.Time{}, timesRequestSent: 0}
+ s.gaps[sk] = append(s.gaps[sk], newG)
+ s.gaps[sk] = mergeGaps(s.gaps[sk])
+ }
+ }
+ newRun := pendingRecordsRange{start: r, end: r, records: map[int64]RecordData{r: record}}
+ s.pending[sk] = addPending(s.pending[sk], newRun)
+
+ var toWrite []RecordData
+ runs := s.pending[sk]
+ for len(runs) > 0 && runs[0].start == s.lastContiguousRowId[sk]+1 {
+ ru := runs[0]
+ for id := ru.start; id <= ru.end; id++ {
+ toWrite = append(toWrite, ru.records[id])
+ }
+ s.lastContiguousRowId[sk] = ru.end
+ s.pending[sk] = runs[1:]
+ runs = s.pending[sk]
+ }
+
+ if len(toWrite) > 0 {
+ s.mu.Lock()
+ s.recordsToWrite = append(s.recordsToWrite, toWrite...)
+ s.mu.Unlock()
+ }
+}
+
+func (s *stateStore) getWriteableMessages() []RecordData {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ records := s.recordsToWrite[:]
+ s.recordsToWrite = []RecordData{}
+ return records
+}
+
+func (s *stateStore) getResendRequests() []ResendRequest {
+ s.mu.RLock()
+ keys := make([]SourceKey, 0, len(s.gaps))
+ for k := range s.gaps {
+ keys = append(keys, k)
+ }
+ s.mu.RUnlock()
+
+ resendRequests := []ResendRequest{}
+ for _, sk := range keys {
+ if _, ok := s.sourceKeyMu[sk]; !ok {
+ continue
+ }
+ s.sourceKeyMu[sk].Lock()
+ gaps, ok := s.gaps[sk]
+ if !ok {
+ s.sourceKeyMu[sk].Unlock()
+ continue
+ }
+ gapRanges := []GapRange{}
+ for i := range gaps {
+ if gaps[i].isResendable() {
+ gapRanges = append(gapRanges, gaps[i].GapRange)
+ gaps[i].lastRequestSent = time.Now()
+ gaps[i].timesRequestSent++
+ }
+ }
+ if len(gapRanges) > 0 {
+ resendRequests = append(resendRequests, ResendRequest{
+ SourceNodeID: sk.SourceNodeId,
+ SourcePath: sk.SourcePath,
+ Gaps: gapRanges,
+ })
+ }
+ s.sourceKeyMu[sk].Unlock()
+ }
+ return resendRequests
+}
+
+func (s *stateStore) getCurrentGaps() map[SourceKey][]gap {
+ s.mu.RLock()
+ defer s.mu.RUnlock()
+ copied := make(map[SourceKey][]gap, len(s.gaps))
+ for k, v := range s.gaps {
+ gapCopy := make([]gap, len(v))
+ copy(gapCopy, v)
+ copied[k] = gapCopy
+ }
+ return copied
+}
+
+func addPending(pending []pendingRecordsRange, newPending pendingRecordsRange) []pendingRecordsRange {
+ temp := append(append([]pendingRecordsRange{}, pending...), newPending)
+ sort.Slice(temp, func(i, j int) bool { return temp[i].start < temp[j].start })
+ merged := []pendingRecordsRange{}
+ for _, p := range temp {
+ if len(merged) == 0 || merged[len(merged)-1].end+1 < p.start {
+ merged = append(merged, p)
+ continue
+ }
+ lastIdx := len(merged) - 1
+ if merged[lastIdx].end < p.end {
+ merged[lastIdx].end = p.end
+ }
+ for k, v := range p.records {
+ merged[lastIdx].records[k] = v
+ }
+ }
+ return merged
+}
+
+func mergeGaps(gaps []gap) []gap {
+ if len(gaps) == 0 {
+ return gaps
+ }
+ sort.Slice(gaps, func(i, j int) bool { return gaps[i].Start < gaps[j].Start })
+ merged := []gap{gaps[0]}
+ for _, g := range gaps[1:] {
+ lastIdx := len(merged) - 1
+ last := merged[lastIdx]
+ if last.End+1 >= g.Start {
+ if last.End < g.End {
+ merged[lastIdx].End = g.End
+ }
+ if g.firstSeen.Before(last.firstSeen) {
+ merged[lastIdx].firstSeen = g.firstSeen
+ }
+ if g.lastRequestSent.After(last.lastRequestSent) {
+ merged[lastIdx].lastRequestSent = g.lastRequestSent
+ }
+ if g.timesRequestSent > last.timesRequestSent {
+ merged[lastIdx].timesRequestSent = g.timesRequestSent
+ }
+ } else {
+ merged = append(merged, g)
+ }
+ }
+ return merged
+}
diff --git a/networking/forwarder/src/state_store_test.go b/networking/forwarder/src/state_store_test.go
new file mode 100644
index 00000000..c0a050f3
--- /dev/null
+++ b/networking/forwarder/src/state_store_test.go
@@ -0,0 +1,283 @@
+package forwarder
+
+import (
+ "testing"
+ "time"
+)
+
+func TestInOrderMessages_SingleSource(t *testing.T) {
+ store := newStateStore(make(map[SourceKey]int64))
+ sk := SourceKey{"node1", "path1"}
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 1,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 2,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 3,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ writeable := store.getWriteableMessages()
+ if len(writeable) != 3 || writeable[0].SourceRowID != 1 || writeable[1].SourceRowID != 2 || writeable[2].SourceRowID != 3 {
+ t.Errorf("Expected 3 contiguous messages, got %v", writeable)
+ }
+
+ gaps := store.getCurrentGaps()[sk]
+ if len(gaps) != 0 {
+ t.Errorf("Expected no gaps, got %v", gaps)
+ }
+
+ if store.lastContiguousRowId[sk] != 3 {
+ t.Errorf("Expected lastContiguous=3, got %d", store.lastContiguousRowId[sk])
+ }
+}
+
+func TestOutOfOrder_CreateGapThenFill(t *testing.T) {
+ store := newStateStore(make(map[SourceKey]int64))
+ sk := SourceKey{"node1", "path1"}
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 1,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 3,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ gaps := store.getCurrentGaps()[sk]
+ if len(gaps) != 1 || gaps[0].Start != 2 || gaps[0].End != 2 {
+ t.Errorf("Expected gap [2,2], got %v", gaps)
+ }
+
+ writeable := store.getWriteableMessages()
+ if len(writeable) != 1 || writeable[0].SourceRowID != 1 {
+ t.Errorf("Expected only 1 written, got %v", writeable)
+ }
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 2,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ writeable = store.getWriteableMessages()
+ if len(writeable) != 2 || writeable[0].SourceRowID != 2 || writeable[1].SourceRowID != 3 {
+ t.Errorf("Expected 1 and 2 written, got %v", writeable)
+ }
+
+ gaps = store.getCurrentGaps()[sk]
+ if len(gaps) != 0 {
+ t.Errorf("Expected no gaps after fill, got %v", gaps)
+ }
+
+ if store.lastContiguousRowId[sk] != 3 {
+ t.Errorf("Expected lastContiguous=3, got %d", store.lastContiguousRowId[sk])
+ }
+}
+
+func TestFillMiddleOfGap_Split(t *testing.T) {
+ store := newStateStore(make(map[SourceKey]int64))
+ sk := SourceKey{"node1", "path1"}
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 1,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 5,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ gaps := store.getCurrentGaps()[sk]
+ if len(gaps) != 1 || gaps[0].Start != 2 || gaps[0].End != 4 {
+ t.Errorf("Expected gap [1,4], got %v", gaps)
+ }
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 3,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ gaps = store.getCurrentGaps()[sk]
+ if len(gaps) != 2 || gaps[0].Start != 2 || gaps[0].End != 2 || gaps[1].Start != 4 || gaps[1].End != 4 {
+ t.Errorf("Expected gaps [1,1] and [3,4], got %v", gaps)
+ }
+
+ writeable := store.getWriteableMessages()
+ if len(writeable) != 1 || writeable[0].SourceRowID != 1 {
+ t.Errorf("Expected only 0 written, got %v", writeable)
+ }
+
+ if len(store.pending[sk]) != 2 {
+ t.Errorf("Expected 2 pending runs, got %d", len(store.pending[sk]))
+ }
+}
+
+func TestMultipleRuns_FillConnectingGap_MergeAndPartialAdvance(t *testing.T) {
+ store := newStateStore(make(map[SourceKey]int64))
+ sk := SourceKey{"node1", "path1"}
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 1,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 2,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 4,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 5,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 7,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ gaps := store.getCurrentGaps()[sk]
+ if len(gaps) != 2 || gaps[0].Start != 3 || gaps[0].End != 3 || gaps[1].Start != 6 || gaps[1].End != 6 {
+ t.Errorf("Expected gaps [3,3],[6,6], got %v", gaps)
+ }
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 3,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ writeable := store.getWriteableMessages()
+ if len(writeable) != 5 || writeable[4].SourceRowID != 5 {
+ t.Errorf("Expected 1-5 written, got %v", writeable)
+ }
+
+ gaps = store.getCurrentGaps()[sk]
+ if len(gaps) != 1 || gaps[0].Start != 6 || gaps[0].End != 6 {
+ t.Errorf("Expected gap [6,6], got %v", gaps)
+ }
+
+ if store.lastContiguousRowId[sk] != 5 {
+ t.Errorf("Expected lastContiguous=5, got %d", store.lastContiguousRowId[sk])
+ }
+
+ if len(store.pending[sk]) != 1 || store.pending[sk][0].start != 7 {
+ t.Errorf("Expected pending [7,7], got %v", store.pending[sk])
+ }
+}
+
+func TestInitialHighRowID_CreateGap_IgnoreDuplicateAndOld(t *testing.T) {
+ store := newStateStore(make(map[SourceKey]int64))
+ sk := SourceKey{"node1", "path1"}
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 3,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ gaps := store.getCurrentGaps()[sk]
+ if len(gaps) != 1 || gaps[0].Start != 1 || gaps[0].End != 2 {
+ t.Errorf("Expected gap [1,2], got %v", gaps)
+ }
+
+ writeable := store.getWriteableMessages()
+ if len(writeable) != 0 {
+ t.Errorf("Expected no writeable, got %v", writeable)
+ }
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: 3,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ if len(store.pending[sk]) != 1 || len(store.pending[sk][0].records) != 1 {
+ t.Errorf("Duplicate added unexpectedly")
+ }
+
+ store.onRecord(RecordData{
+ TrackingData: TrackingData{
+ SourceKey: SourceKey{SourceNodeId: "node1", SourcePath: "path1"},
+ SourceRowID: -1,
+ SourceTimestamp: time.Now(),
+ },
+ Data: nil,
+ })
+
+ if store.lastContiguousRowId[sk] != 0 {
+ t.Errorf("Old message affected lastContiguous")
+ }
+}
diff --git a/networking/.gitignore b/networking/topology/.gitignore
similarity index 100%
rename from networking/.gitignore
rename to networking/topology/.gitignore
diff --git a/networking/Cargo.lock b/networking/topology/Cargo.lock
similarity index 100%
rename from networking/Cargo.lock
rename to networking/topology/Cargo.lock
diff --git a/networking/Cargo.toml b/networking/topology/Cargo.toml
similarity index 100%
rename from networking/Cargo.toml
rename to networking/topology/Cargo.toml
diff --git a/networking/pyproject.toml b/networking/topology/pyproject.toml
similarity index 95%
rename from networking/pyproject.toml
rename to networking/topology/pyproject.toml
index b2f433b7..f2e82e89 100644
--- a/networking/pyproject.toml
+++ b/networking/topology/pyproject.toml
@@ -2,7 +2,6 @@
name = "exo-networking"
version = "0.1.0"
description = "Add your description here"
-readme = "README.md"
authors = [
{ name = "Arbion Halili", email = "99731180+ToxicPine@users.noreply.github.com" }
]
diff --git a/networking/src/lib.rs b/networking/topology/src/lib.rs
similarity index 100%
rename from networking/src/lib.rs
rename to networking/topology/src/lib.rs
diff --git a/networking/src/networking/__init__.py b/networking/topology/src/networking/__init__.py
similarity index 100%
rename from networking/src/networking/__init__.py
rename to networking/topology/src/networking/__init__.py
diff --git a/networking/src/networking/_core.pyi b/networking/topology/src/networking/_core.pyi
similarity index 100%
rename from networking/src/networking/_core.pyi
rename to networking/topology/src/networking/_core.pyi
diff --git a/pyproject.toml b/pyproject.toml
index 6b3c4719..d4573c85 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -37,7 +37,7 @@ members = [
"worker",
"shared",
"engines/*",
- "networking",
+ "networking/topology",
]
[tool.uv.sources]
diff --git a/uv.lock b/uv.lock
index 7bbee8d9..e91fab50 100644
--- a/uv.lock
+++ b/uv.lock
@@ -230,7 +230,7 @@ requires-dist = [
[[package]]
name = "exo-networking"
version = "0.1.0"
-source = { editable = "networking" }
+source = { editable = "networking/topology" }
[[package]]
name = "exo-shared"
@@ -284,7 +284,7 @@ dependencies = [
[package.metadata]
requires-dist = [
{ name = "exo-shared", editable = "shared" },
- { name = "mlx", specifier = ">=0.26.1" },
+ { name = "mlx", specifier = "==0.26.3" },
{ name = "mlx-lm", specifier = ">=0.25.3" },
]
@@ -911,24 +911,24 @@ wheels = [
[[package]]
name = "ruff"
-version = "0.12.3"
+version = "0.12.4"
source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/c3/2a/43955b530c49684d3c38fcda18c43caf91e99204c2a065552528e0552d4f/ruff-0.12.3.tar.gz", hash = "sha256:f1b5a4b6668fd7b7ea3697d8d98857390b40c1320a63a178eee6be0899ea2d77", size = 4459341, upload-time = "2025-07-11T13:21:16.086Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/9b/ce/8d7dbedede481245b489b769d27e2934730791a9a82765cb94566c6e6abd/ruff-0.12.4.tar.gz", hash = "sha256:13efa16df6c6eeb7d0f091abae50f58e9522f3843edb40d56ad52a5a4a4b6873", size = 5131435, upload-time = "2025-07-17T17:27:19.138Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/e2/fd/b44c5115539de0d598d75232a1cc7201430b6891808df111b8b0506aae43/ruff-0.12.3-py3-none-linux_armv6l.whl", hash = "sha256:47552138f7206454eaf0c4fe827e546e9ddac62c2a3d2585ca54d29a890137a2", size = 10430499, upload-time = "2025-07-11T13:20:26.321Z" },
- { url = "https://files.pythonhosted.org/packages/43/c5/9eba4f337970d7f639a37077be067e4ec80a2ad359e4cc6c5b56805cbc66/ruff-0.12.3-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:0a9153b000c6fe169bb307f5bd1b691221c4286c133407b8827c406a55282041", size = 11213413, upload-time = "2025-07-11T13:20:30.017Z" },
- { url = "https://files.pythonhosted.org/packages/e2/2c/fac3016236cf1fe0bdc8e5de4f24c76ce53c6dd9b5f350d902549b7719b2/ruff-0.12.3-py3-none-macosx_11_0_arm64.whl", hash = "sha256:fa6b24600cf3b750e48ddb6057e901dd5b9aa426e316addb2a1af185a7509882", size = 10586941, upload-time = "2025-07-11T13:20:33.046Z" },
- { url = "https://files.pythonhosted.org/packages/c5/0f/41fec224e9dfa49a139f0b402ad6f5d53696ba1800e0f77b279d55210ca9/ruff-0.12.3-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e2506961bf6ead54887ba3562604d69cb430f59b42133d36976421bc8bd45901", size = 10783001, upload-time = "2025-07-11T13:20:35.534Z" },
- { url = "https://files.pythonhosted.org/packages/0d/ca/dd64a9ce56d9ed6cad109606ac014860b1c217c883e93bf61536400ba107/ruff-0.12.3-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:c4faaff1f90cea9d3033cbbcdf1acf5d7fb11d8180758feb31337391691f3df0", size = 10269641, upload-time = "2025-07-11T13:20:38.459Z" },
- { url = "https://files.pythonhosted.org/packages/63/5c/2be545034c6bd5ce5bb740ced3e7014d7916f4c445974be11d2a406d5088/ruff-0.12.3-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:40dced4a79d7c264389de1c59467d5d5cefd79e7e06d1dfa2c75497b5269a5a6", size = 11875059, upload-time = "2025-07-11T13:20:41.517Z" },
- { url = "https://files.pythonhosted.org/packages/8e/d4/a74ef1e801ceb5855e9527dae105eaff136afcb9cc4d2056d44feb0e4792/ruff-0.12.3-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:0262d50ba2767ed0fe212aa7e62112a1dcbfd46b858c5bf7bbd11f326998bafc", size = 12658890, upload-time = "2025-07-11T13:20:44.442Z" },
- { url = "https://files.pythonhosted.org/packages/13/c8/1057916416de02e6d7c9bcd550868a49b72df94e3cca0aeb77457dcd9644/ruff-0.12.3-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:12371aec33e1a3758597c5c631bae9a5286f3c963bdfb4d17acdd2d395406687", size = 12232008, upload-time = "2025-07-11T13:20:47.374Z" },
- { url = "https://files.pythonhosted.org/packages/f5/59/4f7c130cc25220392051fadfe15f63ed70001487eca21d1796db46cbcc04/ruff-0.12.3-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:560f13b6baa49785665276c963edc363f8ad4b4fc910a883e2625bdb14a83a9e", size = 11499096, upload-time = "2025-07-11T13:20:50.348Z" },
- { url = "https://files.pythonhosted.org/packages/d4/01/a0ad24a5d2ed6be03a312e30d32d4e3904bfdbc1cdbe63c47be9d0e82c79/ruff-0.12.3-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:023040a3499f6f974ae9091bcdd0385dd9e9eb4942f231c23c57708147b06311", size = 11688307, upload-time = "2025-07-11T13:20:52.945Z" },
- { url = "https://files.pythonhosted.org/packages/93/72/08f9e826085b1f57c9a0226e48acb27643ff19b61516a34c6cab9d6ff3fa/ruff-0.12.3-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:883d844967bffff5ab28bba1a4d246c1a1b2933f48cb9840f3fdc5111c603b07", size = 10661020, upload-time = "2025-07-11T13:20:55.799Z" },
- { url = "https://files.pythonhosted.org/packages/80/a0/68da1250d12893466c78e54b4a0ff381370a33d848804bb51279367fc688/ruff-0.12.3-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:2120d3aa855ff385e0e562fdee14d564c9675edbe41625c87eeab744a7830d12", size = 10246300, upload-time = "2025-07-11T13:20:58.222Z" },
- { url = "https://files.pythonhosted.org/packages/6a/22/5f0093d556403e04b6fd0984fc0fb32fbb6f6ce116828fd54306a946f444/ruff-0.12.3-py3-none-musllinux_1_2_i686.whl", hash = "sha256:6b16647cbb470eaf4750d27dddc6ebf7758b918887b56d39e9c22cce2049082b", size = 11263119, upload-time = "2025-07-11T13:21:01.503Z" },
- { url = "https://files.pythonhosted.org/packages/92/c9/f4c0b69bdaffb9968ba40dd5fa7df354ae0c73d01f988601d8fac0c639b1/ruff-0.12.3-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:e1417051edb436230023575b149e8ff843a324557fe0a265863b7602df86722f", size = 11746990, upload-time = "2025-07-11T13:21:04.524Z" },
+ { url = "https://files.pythonhosted.org/packages/ae/9f/517bc5f61bad205b7f36684ffa5415c013862dee02f55f38a217bdbe7aa4/ruff-0.12.4-py3-none-linux_armv6l.whl", hash = "sha256:cb0d261dac457ab939aeb247e804125a5d521b21adf27e721895b0d3f83a0d0a", size = 10188824, upload-time = "2025-07-17T17:26:31.412Z" },
+ { url = "https://files.pythonhosted.org/packages/28/83/691baae5a11fbbde91df01c565c650fd17b0eabed259e8b7563de17c6529/ruff-0.12.4-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:55c0f4ca9769408d9b9bac530c30d3e66490bd2beb2d3dae3e4128a1f05c7442", size = 10884521, upload-time = "2025-07-17T17:26:35.084Z" },
+ { url = "https://files.pythonhosted.org/packages/d6/8d/756d780ff4076e6dd035d058fa220345f8c458391f7edfb1c10731eedc75/ruff-0.12.4-py3-none-macosx_11_0_arm64.whl", hash = "sha256:a8224cc3722c9ad9044da7f89c4c1ec452aef2cfe3904365025dd2f51daeae0e", size = 10277653, upload-time = "2025-07-17T17:26:37.897Z" },
+ { url = "https://files.pythonhosted.org/packages/8d/97/8eeee0f48ece153206dce730fc9e0e0ca54fd7f261bb3d99c0a4343a1892/ruff-0.12.4-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e9949d01d64fa3672449a51ddb5d7548b33e130240ad418884ee6efa7a229586", size = 10485993, upload-time = "2025-07-17T17:26:40.68Z" },
+ { url = "https://files.pythonhosted.org/packages/49/b8/22a43d23a1f68df9b88f952616c8508ea6ce4ed4f15353b8168c48b2d7e7/ruff-0.12.4-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:be0593c69df9ad1465e8a2d10e3defd111fdb62dcd5be23ae2c06da77e8fcffb", size = 10022824, upload-time = "2025-07-17T17:26:43.564Z" },
+ { url = "https://files.pythonhosted.org/packages/cd/70/37c234c220366993e8cffcbd6cadbf332bfc848cbd6f45b02bade17e0149/ruff-0.12.4-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:a7dea966bcb55d4ecc4cc3270bccb6f87a337326c9dcd3c07d5b97000dbff41c", size = 11524414, upload-time = "2025-07-17T17:26:46.219Z" },
+ { url = "https://files.pythonhosted.org/packages/14/77/c30f9964f481b5e0e29dd6a1fae1f769ac3fd468eb76fdd5661936edd262/ruff-0.12.4-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:afcfa3ab5ab5dd0e1c39bf286d829e042a15e966b3726eea79528e2e24d8371a", size = 12419216, upload-time = "2025-07-17T17:26:48.883Z" },
+ { url = "https://files.pythonhosted.org/packages/6e/79/af7fe0a4202dce4ef62c5e33fecbed07f0178f5b4dd9c0d2fcff5ab4a47c/ruff-0.12.4-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:c057ce464b1413c926cdb203a0f858cd52f3e73dcb3270a3318d1630f6395bb3", size = 11976756, upload-time = "2025-07-17T17:26:51.754Z" },
+ { url = "https://files.pythonhosted.org/packages/09/d1/33fb1fc00e20a939c305dbe2f80df7c28ba9193f7a85470b982815a2dc6a/ruff-0.12.4-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:e64b90d1122dc2713330350626b10d60818930819623abbb56535c6466cce045", size = 11020019, upload-time = "2025-07-17T17:26:54.265Z" },
+ { url = "https://files.pythonhosted.org/packages/64/f4/e3cd7f7bda646526f09693e2e02bd83d85fff8a8222c52cf9681c0d30843/ruff-0.12.4-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2abc48f3d9667fdc74022380b5c745873499ff827393a636f7a59da1515e7c57", size = 11277890, upload-time = "2025-07-17T17:26:56.914Z" },
+ { url = "https://files.pythonhosted.org/packages/5e/d0/69a85fb8b94501ff1a4f95b7591505e8983f38823da6941eb5b6badb1e3a/ruff-0.12.4-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:2b2449dc0c138d877d629bea151bee8c0ae3b8e9c43f5fcaafcd0c0d0726b184", size = 10348539, upload-time = "2025-07-17T17:26:59.381Z" },
+ { url = "https://files.pythonhosted.org/packages/16/a0/91372d1cb1678f7d42d4893b88c252b01ff1dffcad09ae0c51aa2542275f/ruff-0.12.4-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:56e45bb11f625db55f9b70477062e6a1a04d53628eda7784dce6e0f55fd549eb", size = 10009579, upload-time = "2025-07-17T17:27:02.462Z" },
+ { url = "https://files.pythonhosted.org/packages/23/1b/c4a833e3114d2cc0f677e58f1df6c3b20f62328dbfa710b87a1636a5e8eb/ruff-0.12.4-py3-none-musllinux_1_2_i686.whl", hash = "sha256:478fccdb82ca148a98a9ff43658944f7ab5ec41c3c49d77cd99d44da019371a1", size = 10942982, upload-time = "2025-07-17T17:27:05.343Z" },
+ { url = "https://files.pythonhosted.org/packages/ff/ce/ce85e445cf0a5dd8842f2f0c6f0018eedb164a92bdf3eda51984ffd4d989/ruff-0.12.4-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:0fc426bec2e4e5f4c4f182b9d2ce6a75c85ba9bcdbe5c6f2a74fcb8df437df4b", size = 11343331, upload-time = "2025-07-17T17:27:08.652Z" },
]
[[package]]
@@ -1018,14 +1018,14 @@ wheels = [
[[package]]
name = "starlette"
-version = "0.47.1"
+version = "0.47.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
]
-sdist = { url = "https://files.pythonhosted.org/packages/0a/69/662169fdb92fb96ec3eaee218cf540a629d629c86d7993d9651226a6789b/starlette-0.47.1.tar.gz", hash = "sha256:aef012dd2b6be325ffa16698f9dc533614fb1cebd593a906b90dc1025529a79b", size = 2583072, upload-time = "2025-06-21T04:03:17.337Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/04/57/d062573f391d062710d4088fa1369428c38d51460ab6fedff920efef932e/starlette-0.47.2.tar.gz", hash = "sha256:6ae9aa5db235e4846decc1e7b79c4f346adf41e9777aebeb49dfd09bbd7023d8", size = 2583948, upload-time = "2025-07-20T17:31:58.522Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/82/95/38ef0cd7fa11eaba6a99b3c4f5ac948d8bc6ff199aabd327a29cc000840c/starlette-0.47.1-py3-none-any.whl", hash = "sha256:5e11c9f5c7c3f24959edbf2dffdc01bba860228acf657129467d8a7468591527", size = 72747, upload-time = "2025-06-21T04:03:15.705Z" },
+ { url = "https://files.pythonhosted.org/packages/f7/1f/b876b1f83aef204198a42dc101613fefccb32258e5428b5f9259677864b4/starlette-0.47.2-py3-none-any.whl", hash = "sha256:c5847e96134e5c5371ee9fac6fdf1a67336d5815e09eb2a01fdb57a351ef915b", size = 72984, upload-time = "2025-07-20T17:31:56.738Z" },
]
[[package]]
@@ -1062,7 +1062,7 @@ wheels = [
[[package]]
name = "transformers"
-version = "4.53.2"
+version = "4.53.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "filelock", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
@@ -1076,9 +1076,9 @@ dependencies = [
{ name = "tokenizers", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
{ name = "tqdm", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
]
-sdist = { url = "https://files.pythonhosted.org/packages/4c/67/80f51466ec447028fd84469b208eb742533ce06cc8fad2e3181380199e5c/transformers-4.53.2.tar.gz", hash = "sha256:6c3ed95edfb1cba71c4245758f1b4878c93bf8cde77d076307dacb2cbbd72be2", size = 9201233, upload-time = "2025-07-11T12:39:08.742Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f1/5c/49182918b58eaa0b4c954fd0e37c79fc299e5643e69d70089d0b0eb0cd9b/transformers-4.53.3.tar.gz", hash = "sha256:b2eda1a261de79b78b97f7888fe2005fc0c3fabf5dad33d52cc02983f9f675d8", size = 9197478, upload-time = "2025-07-22T07:30:51.51Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/96/88/beb33a79a382fcd2aed0be5222bdc47f41e4bfe7aaa90ae1374f1d8ea2af/transformers-4.53.2-py3-none-any.whl", hash = "sha256:db8f4819bb34f000029c73c3c557e7d06fc1b8e612ec142eecdae3947a9c78bf", size = 10826609, upload-time = "2025-07-11T12:39:05.461Z" },
+ { url = "https://files.pythonhosted.org/packages/41/b1/d7520cc5cb69c825599042eb3a7c986fa9baa8a8d2dea9acd78e152c81e2/transformers-4.53.3-py3-none-any.whl", hash = "sha256:5aba81c92095806b6baf12df35d756cf23b66c356975fb2a7fa9e536138d7c75", size = 10826382, upload-time = "2025-07-22T07:30:48.458Z" },
]
[[package]]
diff --git a/worker/pyproject.toml b/worker/pyproject.toml
index f1f4871a..49ede7b7 100644
--- a/worker/pyproject.toml
+++ b/worker/pyproject.toml
@@ -6,7 +6,7 @@ readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"exo-shared",
- "mlx>=0.26.1",
+ "mlx==0.26.3",
"mlx-lm>=0.25.3",
]
← 53c652c3 Fix tests!
·
back to Exo
·
New API! 14b3c4a6 →