chore(release): 0.1.3

chore(release): 0.1.2

Co-authored-by: cuddle-please <bot@cuddle.sh>
Reviewed-on: https://git.front.kjuulh.io/kjuulh/noleader/pulls/2

.env

@@ -0,0 +1,3 @@
+DATABASE_URL=postgres://devuser:devpassword@localhost:5432/dev
+
+#SQLX_OFFLINE=true

.sqlx/query-0461b6433be16583d2480de11d5b712de1229dff78624ecab5edcf9f05a2e0e4.json

@@ -0,0 +1,40 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            SELECT value, revision\n            FROM noleader_leaders\n            WHERE\n                  key = $1\n              AND heartbeat >= now() - interval '60 seconds'\n            ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value",
+        "type_info": "Text",
+        "origin": {
+          "Table": {
+            "table": "noleader_leaders",
+            "name": "value"
+          }
+        }
+      },
+      {
+        "ordinal": 1,
+        "name": "revision",
+        "type_info": "Int8",
+        "origin": {
+          "Table": {
+            "table": "noleader_leaders",
+            "name": "revision"
+          }
+        }
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "0461b6433be16583d2480de11d5b712de1229dff78624ecab5edcf9f05a2e0e4"
+}

.sqlx/query-1aa9d51fee3918db168e3704d1ac0e80e5038e2619e5029597fd28d4967538c2.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n                DELETE FROM noleader_leaders\n                WHERE\n                        key = $1\n                    AND value = $2\n                    AND revision = $3\n            ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "1aa9d51fee3918db168e3704d1ac0e80e5038e2619e5029597fd28d4967538c2"
+}

.sqlx/query-976e720a4dee2911278524a199d6d5ece23f141e4b4c094efe5fc3123e376b04.json

@@ -0,0 +1,43 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            INSERT INTO noleader_leaders (key, value, revision, heartbeat)\n            VALUES ($1, $2, $3, now())\n            ON CONFLICT (key)\n            DO UPDATE SET\n                value = EXCLUDED.value,\n                revision = EXCLUDED.revision,\n                heartbeat = now()\n            WHERE \n                (\n                    -- Normal case: revision matches (we're the current leader updating)\n                    noleader_leaders.revision = $4\n                    OR\n                    -- Override case: heartbeat is old (stale leader)\n                    noleader_leaders.heartbeat < now() - INTERVAL '60 seconds'\n                )\n            RETURNING value, revision\n            ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value",
+        "type_info": "Text",
+        "origin": {
+          "Table": {
+            "table": "noleader_leaders",
+            "name": "value"
+          }
+        }
+      },
+      {
+        "ordinal": 1,
+        "name": "revision",
+        "type_info": "Int8",
+        "origin": {
+          "Table": {
+            "table": "noleader_leaders",
+            "name": "revision"
+          }
+        }
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "976e720a4dee2911278524a199d6d5ece23f141e4b4c094efe5fc3123e376b04"
+}

CHANGELOG.md

@@ -0,0 +1,29 @@
+# Changelog
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [Unreleased]
+
+## [0.1.3] - 2025-09-24
+
+### Added
+- add features for nats and postgres
+- add postgres
+- extract backend
+- do publish
+- allow readme
+
+## [0.1.2] - 2025-07-04
+
+### Added
+- update basic example with a more simple acquire and run function
+
+### Docs
+- update master
+
+## [0.1.1] - 2025-07-04
+
+### Added
+- empty commit

Cargo.lock

@@ -26,6 +26,12 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "allocator-api2"
+version = "0.2.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
+
 [[package]]
 name = "anyhow"
 version = "1.0.98"
@@ -57,7 +63,7 @@ dependencies = [
  "serde_json",
  "serde_nanos",
  "serde_repr",
- "thiserror",
+ "thiserror 1.0.69",
  "time",
  "tokio",
  "tokio-rustls",
@@ -68,6 +74,26 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "async-trait"
+version = "0.1.89"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "atoi"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528"
+dependencies = [
+ "num-traits",
+]
+
 [[package]]
 name = "autocfg"
 version = "1.5.0"
@@ -106,6 +132,9 @@ name = "bitflags"
 version = "2.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1b8e56985ec62d17e9c1001dc89c88ecd7dc08e47eba5ec7c29c7b5eeecde967"
+dependencies = [
+ "serde",
+]
 
 [[package]]
 name = "block-buffer"
@@ -122,6 +151,12 @@ version = "3.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43"
 
+[[package]]
+name = "byteorder"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
+
 [[package]]
 name = "bytes"
 version = "1.10.1"
@@ -146,6 +181,15 @@ version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9555578bc9e57714c812a1f84e4fc5b4d21fcb063490c624de019f7464c91268"
 
+[[package]]
+name = "concurrent-queue"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "const-oid"
 version = "0.9.6"
@@ -177,6 +221,36 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "crc"
+version = "3.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9710d3b3739c2e349eb44fe848ad0b7c8cb1e42bd87ee49371df2f7acaf3e675"
+dependencies = [
+ "crc-catalog",
+]
+
+[[package]]
+name = "crc-catalog"
+version = "2.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5"
+
+[[package]]
+name = "crossbeam-queue"
+version = "0.3.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115"
+dependencies = [
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-utils"
+version = "0.8.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28"
+
 [[package]]
 name = "crypto-common"
 version = "0.1.6"
@@ -247,7 +321,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
  "block-buffer",
+ "const-oid",
  "crypto-common",
+ "subtle",
 ]
 
 [[package]]
@@ -261,6 +337,12 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "dotenvy"
+version = "0.15.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b"
+
 [[package]]
 name = "ed25519"
 version = "2.2.3"
@@ -283,18 +365,72 @@ dependencies = [
  "subtle",
 ]
 
+[[package]]
+name = "either"
+version = "1.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
+dependencies = [
+ "serde",
+]
+
+[[package]]
+name = "equivalent"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+
+[[package]]
+name = "etcetera"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26c7b13d0780cb82722fd59f6f57f925e143427e4a75313a6c77243bf5326ae6"
+dependencies = [
+ "cfg-if",
+ "home",
+ "windows-sys 0.59.0",
+]
+
+[[package]]
+name = "event-listener"
+version = "5.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab"
+dependencies = [
+ "concurrent-queue",
+ "parking",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "fiat-crypto"
 version = "0.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d"
 
+[[package]]
+name = "flume"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "spin",
+]
+
 [[package]]
 name = "fnv"
 version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
 
+[[package]]
+name = "foldhash"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
+
 [[package]]
 name = "form_urlencoded"
 version = "1.2.1"
@@ -334,6 +470,28 @@ version = "0.3.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
 
+[[package]]
+name = "futures-executor"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-intrusive"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f"
+dependencies = [
+ "futures-core",
+ "lock_api",
+ "parking_lot",
+]
+
 [[package]]
 name = "futures-io"
 version = "0.3.31"
@@ -408,6 +566,71 @@ version = "0.31.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f"
 
+[[package]]
+name = "hashbrown"
+version = "0.15.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
+dependencies = [
+ "allocator-api2",
+ "equivalent",
+ "foldhash",
+]
+
+[[package]]
+name = "hashbrown"
+version = "0.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5419bdc4f6a9207fbeba6d11b604d481addf78ecd10c11ad51e76c2f6482748d"
+
+[[package]]
+name = "hashlink"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1"
+dependencies = [
+ "hashbrown 0.15.5",
+]
+
+[[package]]
+name = "heck"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
+
+[[package]]
+name = "hex"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
+
+[[package]]
+name = "hkdf"
+version = "0.12.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7"
+dependencies = [
+ "hmac",
+]
+
+[[package]]
+name = "hmac"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
+dependencies = [
+ "digest",
+]
+
+[[package]]
+name = "home"
+version = "0.5.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "589533453244b0995c858700322199b2becb13b627df2851f64a2775d024abcf"
+dependencies = [
+ "windows-sys 0.59.0",
+]
+
 [[package]]
 name = "http"
 version = "1.3.1"
@@ -532,6 +755,16 @@ dependencies = [
  "icu_properties",
 ]
 
+[[package]]
+name = "indexmap"
+version = "2.11.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b0f83760fb341a774ed326568e19f5a863af4a952def8c39f9ab92fd95b88e5"
+dependencies = [
+ "equivalent",
+ "hashbrown 0.16.0",
+]
+
 [[package]]
 name = "io-uring"
 version = "0.7.8"
@@ -564,6 +797,9 @@ name = "lazy_static"
 version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
+dependencies = [
+ "spin",
+]
 
 [[package]]
 name = "libc"
@@ -571,6 +807,33 @@ version = "0.2.174"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1171693293099992e19cddea4e8b849964e9846f4acee11b3948bcc337be8776"
 
+[[package]]
+name = "libm"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f9fbbcab51052fe104eb5e5d351cf728d30a5be1fe14d9be8a3b097481fb97de"
+
+[[package]]
+name = "libredox"
+version = "0.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "416f7e718bdb06000964960ffa43b4335ad4012ae8b99060261aa4a8088d5ccb"
+dependencies = [
+ "bitflags",
+ "libc",
+ "redox_syscall",
+]
+
+[[package]]
+name = "libsqlite3-sys"
+version = "0.30.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149"
+dependencies = [
+ "pkg-config",
+ "vcpkg",
+]
+
 [[package]]
 name = "litemap"
 version = "0.8.0"
@@ -602,6 +865,16 @@ dependencies = [
  "regex-automata 0.1.10",
 ]
 
+[[package]]
+name = "md-5"
+version = "0.10.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d89e7ee0cfbedfc4da3340218492196241d89eefb6dab27de5df917a6d2e78cf"
+dependencies = [
+ "cfg-if",
+ "digest",
+]
+
 [[package]]
 name = "memchr"
 version = "2.7.5"
@@ -645,12 +918,14 @@ dependencies = [
 
 [[package]]
 name = "noleader"
-version = "0.1.0"
+version = "0.1.2"
 dependencies = [
  "anyhow",
  "async-nats",
+ "async-trait",
  "bytes",
  "rand 0.9.1",
+ "sqlx",
  "tokio",
  "tokio-util",
  "tracing",
@@ -677,12 +952,59 @@ dependencies = [
  "rand 0.8.5",
 ]
 
+[[package]]
+name = "num-bigint-dig"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc84195820f291c7697304f3cbdadd1cb7199c0efc917ff5eafd71225c136151"
+dependencies = [
+ "byteorder",
+ "lazy_static",
+ "libm",
+ "num-integer",
+ "num-iter",
+ "num-traits",
+ "rand 0.8.5",
+ "smallvec",
+ "zeroize",
+]
+
 [[package]]
 name = "num-conv"
 version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9"
 
+[[package]]
+name = "num-integer"
+version = "0.1.46"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f"
+dependencies = [
+ "num-traits",
+]
+
+[[package]]
+name = "num-iter"
+version = "0.1.45"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1429034a0490724d0075ebb2bc9e875d6503c3cf69e235a8941aa757d83ef5bf"
+dependencies = [
+ "autocfg",
+ "num-integer",
+ "num-traits",
+]
+
+[[package]]
+name = "num-traits"
+version = "0.2.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841"
+dependencies = [
+ "autocfg",
+ "libm",
+]
+
 [[package]]
 name = "object"
 version = "0.36.7"
@@ -710,6 +1032,12 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
 
+[[package]]
+name = "parking"
+version = "2.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
+
 [[package]]
 name = "parking_lot"
 version = "0.12.4"
@@ -780,6 +1108,17 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
+[[package]]
+name = "pkcs1"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f"
+dependencies = [
+ "der",
+ "pkcs8",
+ "spki",
+]
+
 [[package]]
 name = "pkcs8"
 version = "0.10.2"
@@ -790,6 +1129,12 @@ dependencies = [
  "spki",
 ]
 
+[[package]]
+name = "pkg-config"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
+
 [[package]]
 name = "portable-atomic"
 version = "1.11.1"
@@ -905,9 +1250,9 @@ dependencies = [
 
 [[package]]
 name = "redox_syscall"
-version = "0.5.13"
+version = "0.5.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0d04b7d0ee6b4a0207a0a7adb104d23ecb0b47d6beae7152d0fa34b692b29fd6"
+checksum = "5407465600fb0548f1442edf71dd20683c6ed326200ace4b1ef0763521bb3b77"
 dependencies = [
  "bitflags",
 ]
@@ -970,6 +1315,26 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
+[[package]]
+name = "rsa"
+version = "0.9.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "78928ac1ed176a5ca1d17e578a1825f3d81ca54cf41053a592584b020cfd691b"
+dependencies = [
+ "const-oid",
+ "digest",
+ "num-bigint-dig",
+ "num-integer",
+ "num-traits",
+ "pkcs1",
+ "pkcs8",
+ "rand_core 0.6.4",
+ "signature",
+ "spki",
+ "subtle",
+ "zeroize",
+]
+
 [[package]]
 name = "rustc-demangle"
 version = "0.1.25"
@@ -1159,6 +1524,29 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "serde_urlencoded"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd"
+dependencies = [
+ "form_urlencoded",
+ "itoa",
+ "ryu",
+ "serde",
+]
+
+[[package]]
+name = "sha1"
+version = "0.10.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "digest",
+]
+
 [[package]]
 name = "sha2"
 version = "0.10.9"
@@ -1227,6 +1615,9 @@ name = "smallvec"
 version = "1.15.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
+dependencies = [
+ "serde",
+]
 
 [[package]]
 name = "socket2"
@@ -1238,6 +1629,15 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
+[[package]]
+name = "spin"
+version = "0.9.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
+dependencies = [
+ "lock_api",
+]
+
 [[package]]
 name = "spki"
 version = "0.7.3"
@@ -1248,12 +1648,208 @@ dependencies = [
  "der",
 ]
 
+[[package]]
+name = "sqlx"
+version = "0.9.0-alpha.1"
+source = "git+https://github.com/launchbadge/sqlx?rev=064d649abdfd1742e5fdcc20176a6b415b9c25d3#064d649abdfd1742e5fdcc20176a6b415b9c25d3"
+dependencies = [
+ "sqlx-core",
+ "sqlx-macros",
+ "sqlx-mysql",
+ "sqlx-postgres",
+ "sqlx-sqlite",
+]
+
+[[package]]
+name = "sqlx-core"
+version = "0.9.0-alpha.1"
+source = "git+https://github.com/launchbadge/sqlx?rev=064d649abdfd1742e5fdcc20176a6b415b9c25d3#064d649abdfd1742e5fdcc20176a6b415b9c25d3"
+dependencies = [
+ "base64",
+ "bytes",
+ "cfg-if",
+ "crc",
+ "crossbeam-queue",
+ "either",
+ "event-listener",
+ "futures-core",
+ "futures-intrusive",
+ "futures-io",
+ "futures-util",
+ "hashbrown 0.15.5",
+ "hashlink",
+ "indexmap",
+ "log",
+ "memchr",
+ "percent-encoding",
+ "rustls",
+ "serde",
+ "serde_json",
+ "sha2",
+ "smallvec",
+ "thiserror 2.0.16",
+ "tokio",
+ "tokio-stream",
+ "tracing",
+ "url",
+ "uuid",
+ "webpki-roots 0.26.11",
+]
+
+[[package]]
+name = "sqlx-macros"
+version = "0.9.0-alpha.1"
+source = "git+https://github.com/launchbadge/sqlx?rev=064d649abdfd1742e5fdcc20176a6b415b9c25d3#064d649abdfd1742e5fdcc20176a6b415b9c25d3"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "sqlx-core",
+ "sqlx-macros-core",
+ "syn",
+]
+
+[[package]]
+name = "sqlx-macros-core"
+version = "0.9.0-alpha.1"
+source = "git+https://github.com/launchbadge/sqlx?rev=064d649abdfd1742e5fdcc20176a6b415b9c25d3#064d649abdfd1742e5fdcc20176a6b415b9c25d3"
+dependencies = [
+ "cfg-if",
+ "dotenvy",
+ "either",
+ "heck",
+ "hex",
+ "proc-macro2",
+ "quote",
+ "serde",
+ "serde_json",
+ "sha2",
+ "sqlx-core",
+ "sqlx-mysql",
+ "sqlx-postgres",
+ "sqlx-sqlite",
+ "syn",
+ "tokio",
+ "url",
+]
+
+[[package]]
+name = "sqlx-mysql"
+version = "0.9.0-alpha.1"
+source = "git+https://github.com/launchbadge/sqlx?rev=064d649abdfd1742e5fdcc20176a6b415b9c25d3#064d649abdfd1742e5fdcc20176a6b415b9c25d3"
+dependencies = [
+ "atoi",
+ "base64",
+ "bitflags",
+ "byteorder",
+ "bytes",
+ "crc",
+ "digest",
+ "dotenvy",
+ "either",
+ "futures-channel",
+ "futures-core",
+ "futures-io",
+ "futures-util",
+ "generic-array",
+ "hex",
+ "hkdf",
+ "hmac",
+ "itoa",
+ "log",
+ "md-5",
+ "memchr",
+ "percent-encoding",
+ "rand 0.8.5",
+ "rsa",
+ "serde",
+ "sha1",
+ "sha2",
+ "smallvec",
+ "sqlx-core",
+ "stringprep",
+ "thiserror 2.0.16",
+ "tracing",
+ "uuid",
+ "whoami",
+]
+
+[[package]]
+name = "sqlx-postgres"
+version = "0.9.0-alpha.1"
+source = "git+https://github.com/launchbadge/sqlx?rev=064d649abdfd1742e5fdcc20176a6b415b9c25d3#064d649abdfd1742e5fdcc20176a6b415b9c25d3"
+dependencies = [
+ "atoi",
+ "base64",
+ "bitflags",
+ "byteorder",
+ "crc",
+ "dotenvy",
+ "etcetera",
+ "futures-channel",
+ "futures-core",
+ "futures-util",
+ "hex",
+ "hkdf",
+ "hmac",
+ "home",
+ "itoa",
+ "log",
+ "md-5",
+ "memchr",
+ "rand 0.8.5",
+ "serde",
+ "serde_json",
+ "sha2",
+ "smallvec",
+ "sqlx-core",
+ "stringprep",
+ "thiserror 2.0.16",
+ "tracing",
+ "uuid",
+ "whoami",
+]
+
+[[package]]
+name = "sqlx-sqlite"
+version = "0.9.0-alpha.1"
+source = "git+https://github.com/launchbadge/sqlx?rev=064d649abdfd1742e5fdcc20176a6b415b9c25d3#064d649abdfd1742e5fdcc20176a6b415b9c25d3"
+dependencies = [
+ "atoi",
+ "flume",
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-intrusive",
+ "futures-util",
+ "libsqlite3-sys",
+ "log",
+ "percent-encoding",
+ "serde",
+ "serde_urlencoded",
+ "sqlx-core",
+ "thiserror 2.0.16",
+ "tracing",
+ "url",
+ "uuid",
+]
+
 [[package]]
 name = "stable_deref_trait"
 version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
 
+[[package]]
+name = "stringprep"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1"
+dependencies = [
+ "unicode-bidi",
+ "unicode-normalization",
+ "unicode-properties",
+]
+
 [[package]]
 name = "subtle"
 version = "2.6.1"
@@ -1288,7 +1884,16 @@ version = "1.0.69"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
 dependencies = [
- "thiserror-impl",
+ "thiserror-impl 1.0.69",
+]
+
+[[package]]
+name = "thiserror"
+version = "2.0.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3467d614147380f2e4e374161426ff399c91084acd2363eaf549172b3d5e60c0"
+dependencies = [
+ "thiserror-impl 2.0.16",
 ]
 
 [[package]]
@@ -1302,6 +1907,17 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "thiserror-impl"
+version = "2.0.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c5e1be1c48b9172ee610da68fd9cd2770e7a4056cb3fc98710ee6906f0c7960"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "thread_local"
 version = "1.1.9"
@@ -1352,6 +1968,21 @@ dependencies = [
  "zerovec",
 ]
 
+[[package]]
+name = "tinyvec"
+version = "1.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bfa5fdc3bce6191a1dbc8c02d5c8bffcf557bafa17c124c5264a458f1b0613fa"
+dependencies = [
+ "tinyvec_macros",
+]
+
+[[package]]
+name = "tinyvec_macros"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
+
 [[package]]
 name = "tokio"
 version = "1.46.0"
@@ -1393,6 +2024,17 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-stream"
+version = "0.1.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eca58d7bba4a75707817a2c44174253f9236b2d5fbd055602e9d5c07c139a047"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.15"
@@ -1505,12 +2147,33 @@ version = "1.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1dccffe3ce07af9386bfd29e80c0ab1a8205a2fc34e4bcd40364df902cfa8f3f"
 
+[[package]]
+name = "unicode-bidi"
+version = "0.3.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5"
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5a5f39404a5da50712a4c1eecf25e90dd62b613502b7e925fd4e4d19b5c96512"
 
+[[package]]
+name = "unicode-normalization"
+version = "0.1.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5033c97c4262335cded6d6fc3e5c18ab755e1a3dc96376350f3d8e9f009ad956"
+dependencies = [
+ "tinyvec",
+]
+
+[[package]]
+name = "unicode-properties"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e70f2a8b45122e719eb623c01822704c4e0907e7e426a05927e1a1cfff5b75d0"
+
 [[package]]
 name = "untrusted"
 version = "0.9.0"
@@ -1551,6 +2214,12 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
 
+[[package]]
+name = "vcpkg"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
+
 [[package]]
 name = "version_check"
 version = "0.9.5"
@@ -1572,6 +2241,12 @@ dependencies = [
  "wit-bindgen-rt",
 ]
 
+[[package]]
+name = "wasite"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b"
+
 [[package]]
 name = "wasm-bindgen"
 version = "0.2.100"
@@ -1648,6 +2323,16 @@ dependencies = [
  "rustls-pki-types",
 ]
 
+[[package]]
+name = "whoami"
+version = "1.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d4a4db5077702ca3015d3d02d74974948aba2ad9e12ab7df718ee64ccd7e97d"
+dependencies = [
+ "libredox",
+ "wasite",
+]
+
 [[package]]
 name = "winapi"
 version = "0.3.9"

Cargo.toml

@@ -3,7 +3,7 @@ members = ["crates/*"]
 resolver = "2"
 
 [workspace.package]
-version = "0.0.1"
+version = "0.1.3"
 license = "MIT"
 
 [workspace.dependencies]

README.md

@@ -22,9 +22,9 @@ This library is still young and the API is subject to change.
 
 ## Intended use-case
 
-Noleader is not built for distributed consensus, or fast re-election produces. It take upwards to a minute to get reelected, state is the users responsibility to handle.
+Noleader is not built for distributed consensus, or fast re-election procedures. It take upwards to a minute to get re-elected, state is the users responsibility to handle.
 
-Noleader is pretty much just a distributed lock, intended for use-cases where the use wants to only have a single node scheduling work etc.
+Noleader is pretty much just a distributed lock, intended for use-cases where the user wants to only have a single node scheduling work etc.
 
 Good alternatives are:
 
@@ -71,58 +71,40 @@ async fn main() -> anyhow::Result<()> {
     // Ensure the KV bucket exists
     leader.create_bucket().await?;
 
-    // Spawn the election loop
+    // Attempts to acquire election loop, will call inner function if it wins, if it loses it will retry over again.
-    tokio::spawn({
+    // Will block until either the inner function returns and error, or the election processes crashes, intended to allow the application to properly restart
-        let leader = leader.clone();
-        async move {
     leader
-                .start(CancellationToken::default())
+        .acquire_and_run({
-                .await
+            move |token| {
-                .expect("leadership loop failed");
+                let leader_id = leader_id.clone();
-        }
-    });
 
-    // Do work while we are the leader
+                async move {
-    leader
-        .do_while_leader(|cancel_token| async move {
                     loop {
-                if cancel_token.is_cancelled() {
+                        if token.is_cancelled() {
-                    break;
+                            return Ok(());
                         }
-                tracing::info!("🔑 I am the leader—doing work");
+
+                        tracing::info!(leader_id, "do work as leader");
                         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                     }
-            Ok(())
+                }
+            }
         })
         .await?;
 
+
     Ok(())
 }
 ```
 
-## API Overview
+## Examples
 
-* **`Leader::new(bucket: &str, key: &str, client: async_nats::Client) -> Leader`**
+See the examples folder in ./crates/noleader/examples
-  Create a new election participant.
-* **`create_bucket(&self) -> anyhow::Result<()>`**
-  Ensures the KV bucket exists (no-op if already created).
-* **`start(&self, token: CancellationToken) -> anyhow::Result<()>`**
-  Begins the background leader-election loop; renews TTL on success or retries on failure.
-* **`do_while_leader<F, Fut>(&self, f: F) -> anyhow::Result<()>`**
-  Runs your closure as long as you hold leadership; cancels immediately on loss.
-* **`leader_id(&self) -> Uuid`**
-  Returns your unique candidate ID.
-* **`is_leader(&self) -> Status`**
-  Returns `Status::Leader` or `Status::Candidate`, taking shutdown into account.
 
-### Types
+## Architecture
+
+Noleader uses a simple election stealing 
 
-```rust
-pub enum Status {
-    Leader,
-    Candidate,
-}
-```
 
 ## License
 

crates/noleader/Cargo.toml

@@ -1,18 +1,39 @@
 [package]
 name = "noleader"
-version = "0.1.0"
+edition = "2024"
-edition = "2021"
+readme = "../../README.md"
+version.workspace = true
+license.workspace = true
+repository = "https://git.front.kjuulh.io/kjuulh/noleader"
+authors = ["kjuulh <contact@kasperhermansen.com>"]
+description = "A small leader election package using NATS/Postgres keyvalue store as the distributed locking mechanism. Does not require a min / max set of nodes"
 
 [dependencies]
 anyhow.workspace = true
 tracing.workspace = true
-
-async-nats = "0.42"
-uuid = { version = "1", features = ["v4"] }
-bytes = "1"
 tokio.workspace = true
+
+uuid = { version = "1", features = ["v4", "v7"] }
+bytes = "1"
 tokio-util = "0.7"
-rand = "0.9.1"
+rand = "0.9"
+async-trait = "0.1"
+
+async-nats = { version = "0.42", optional = true }
+
+# fork until dangerous set migrate table name is stable. Should be any version after 8.6
+sqlx = { git = "https://github.com/launchbadge/sqlx", features = [
+  "uuid",
+  "postgres",
+  "runtime-tokio",
+  "tls-rustls",
+], rev = "064d649abdfd1742e5fdcc20176a6b415b9c25d3", optional = true }
 
 [dev-dependencies]
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+
+
+[features]
+default = ["nats", "postgres"]
+nats = ["dep:async-nats"]
+postgres = ["dep:sqlx"]

crates/noleader/examples/basic/main.rs

@@ -1,4 +1,3 @@
-use tokio_util::sync::CancellationToken;
 use tracing_subscriber::EnvFilter;
 
 #[tokio::main]
@@ -14,38 +13,28 @@ async fn main() -> anyhow::Result<()> {
         .init();
 
     let mybucket = "mytestbucket";
-    let mykey = "myleaderkey";
+    let mykey = "basic";
     let client = async_nats::connect("localhost:4222").await?;
 
-    let leader = noleader::Leader::new(mybucket, mykey, client);
+    let leader = noleader::Leader::new_nats(mykey, mybucket, client);
     let leader_id = leader.leader_id().await.to_string();
 
-    tracing::info!("creating bucket");
+    leader
-    leader.create_bucket().await?;
+        .acquire_and_run({
-
+            move |token| {
-    tokio::spawn({
-        let leader = leader.clone();
                 let leader_id = leader_id.clone();
 
                 async move {
-            tracing::debug!(leader_id, "starting leader");
-            leader
-                .start(CancellationToken::default())
-                .await
-                .expect("to succeed");
-        }
-    });
-
-    leader
-        .do_while_leader(move |token| async move {
                     loop {
                         if token.is_cancelled() {
                             return Ok(());
                         }
 
-                tracing::info!("do work as leader");
+                        tracing::info!(leader_id, "do work as leader");
                         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                     }
+                }
+            }
         })
         .await?;
 

crates/noleader/examples/lots_of_candidates/main.rs

@@ -1,4 +1,3 @@
-use tokio_util::sync::CancellationToken;
 use tracing_subscriber::EnvFilter;
 
 #[tokio::main]
@@ -23,22 +22,16 @@ async fn main() -> anyhow::Result<()> {
         let client = client.clone();
 
         let handle = tokio::spawn(async move {
-            let leader = noleader::Leader::new(mybucket, mykey, client);
+            let leader = noleader::Leader::new_nats(mykey, mybucket, client);
             let leader_id = leader.leader_id().await.to_string();
 
-            tracing::info!("creating bucket");
-            leader.create_bucket().await?;
-
             tokio::spawn({
                 let leader = leader.clone();
                 let leader_id = leader_id.clone();
 
                 async move {
                     tracing::debug!(leader_id, "starting leader");
-                    leader
+                    leader.start().await.expect("to succeed");
-                        .start(CancellationToken::default())
-                        .await
-                        .expect("to succeed");
                 }
             });
 

crates/noleader/examples/lots_of_postgres/main.rs

@@ -0,0 +1,96 @@
+use anyhow::Context;
+use tokio_util::sync::CancellationToken;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    // Set up logger
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::from_default_env()
+                .add_directive("noleader=debug".parse().unwrap())
+                .add_directive("lots_of_candidates=debug".parse().unwrap())
+                .add_directive("info".parse().unwrap()),
+        )
+        .init();
+
+    let mykey = "myleaderkey";
+
+    let mut handles = Vec::new();
+
+    let db_url = &std::env::var("DATABASE_URL").context("DATABASE_URL is missing")?;
+    let pool = sqlx::PgPool::connect_lazy(db_url)?;
+
+    let cancel = CancellationToken::new();
+    let mut cancelled_resp = Vec::new();
+
+    tokio::spawn({
+        let cancel = cancel.clone();
+
+        async move {
+            tokio::signal::ctrl_c().await.expect("to receive shutdown");
+
+            cancel.cancel();
+        }
+    });
+
+    for _ in 0..100 {
+        let pool = pool.clone();
+        let cancel = cancel.child_token();
+
+        let item_cancellation = CancellationToken::new();
+        cancelled_resp.push(item_cancellation.child_token());
+
+        let handle = tokio::spawn(async move {
+            let mut leader = noleader::Leader::new_postgres_pool(mykey, pool);
+
+            leader.with_cancellation(cancel);
+            let leader_id = leader.leader_id().await.to_string();
+
+            tokio::spawn({
+                let leader = leader.clone();
+                let leader_id = leader_id.clone();
+
+                async move {
+                    tracing::debug!(leader_id, "starting leader");
+                    let res = leader.start().await;
+
+                    tracing::warn!("shutting down");
+
+                    item_cancellation.cancel();
+
+                    if let Err(e) = res {
+                        tracing::error!("lots failed: {e:?}");
+                    }
+                }
+            });
+
+            loop {
+                tokio::time::sleep(std::time::Duration::from_millis(10000)).await;
+                match leader.is_leader().await {
+                    noleader::Status::Leader => {
+                        tracing::info!(leader_id, "is leader");
+                    }
+                    noleader::Status::Candidate => {
+                        //tracing::debug!("is candiate");
+                    }
+                }
+            }
+
+            #[allow(unreachable_code)]
+            Ok::<(), anyhow::Error>(())
+        });
+
+        handles.push(handle);
+    }
+
+    for cancel in cancelled_resp {
+        cancel.cancelled().await;
+    }
+
+    for handle in handles {
+        handle.abort();
+    }
+
+    Ok(())
+}

crates/noleader/examples/postgres/main.rs

@@ -0,0 +1,49 @@
+use anyhow::Context;
+use tokio::signal;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    // Set up logger
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::from_default_env()
+                .add_directive("noleader=debug".parse().unwrap())
+                .add_directive("lots_of_candidates=debug".parse().unwrap())
+                .add_directive("info".parse().unwrap()),
+        )
+        .init();
+
+    let mykey = "postgres";
+
+    let mut leader = noleader::Leader::new_postgres(
+        mykey,
+        &std::env::var("DATABASE_URL").context("DATABASE_URL is missing")?,
+    );
+    leader.with_cancel_task(async move {
+        signal::ctrl_c().await.unwrap();
+    });
+
+    let leader_id = leader.leader_id().await.to_string();
+
+    leader
+        .acquire_and_run({
+            move |token| {
+                let leader_id = leader_id.clone();
+
+                async move {
+                    loop {
+                        if token.is_cancelled() {
+                            return Ok(());
+                        }
+
+                        tracing::info!(leader_id, "do work as leader");
+                        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                    }
+                }
+            }
+        })
+        .await?;
+
+    Ok(())
+}

crates/noleader/migrations/postgres/20250924114540_initial.sql

@@ -0,0 +1,8 @@
+-- Add migration script here
+
+CREATE TABLE IF NOT EXISTS noleader_leaders (
+    key TEXT PRIMARY KEY NOT NULL,
+    value TEXT NOT NULL,
+    revision BIGINT NOT NULL,
+    heartbeat TIMESTAMPTZ NOT NULL DEFAULT now()
+);

crates/noleader/src/backend.rs

@@ -0,0 +1,116 @@
+use std::{ops::Deref, sync::Arc};
+
+
+    #[cfg(feature = "nats")]
+mod nats;
+    #[cfg(feature = "postgres")]
+mod postgres;
+
+pub struct Backend {
+    inner: Arc<dyn BackendEdge + Send + Sync + 'static>,
+}
+
+impl Backend {
+    pub fn new(edge: impl BackendEdge + Send + Sync + 'static) -> Self {
+        Self {
+            inner: Arc::new(edge),
+        }
+    }
+
+    #[cfg(feature = "nats")]
+    pub fn nats(client: async_nats::Client, bucket: &str) -> Self {
+        Self {
+            inner: Arc::new(nats::NatsBackend::new(client, bucket)),
+        }
+    }
+
+    #[cfg(feature = "postgres")]
+    pub fn postgres(database_url: &str) -> Self {
+        Self {
+            inner: Arc::new(postgres::PostgresBackend::new(database_url)),
+        }
+    }
+
+    #[cfg(feature = "postgres")]
+    pub fn postgres_with_pool(pool: sqlx::PgPool) -> Self {
+        Self {
+            inner: Arc::new(postgres::PostgresBackend::new_with_pool("bogus", pool)),
+        }
+    }
+}
+
+impl Deref for Backend {
+    type Target = Arc<dyn BackendEdge + Send + Sync + 'static>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+#[async_trait::async_trait]
+pub trait BackendEdge {
+    async fn setup(&self) -> anyhow::Result<()>;
+    async fn get(&self, key: &Key) -> anyhow::Result<LeaderValue>;
+    async fn update(&self, key: &Key, val: &LeaderId) -> anyhow::Result<()>;
+    async fn release(&self, key: &Key, val: &LeaderId) -> anyhow::Result<()>;
+}
+
+pub enum LeaderValue {
+    Unknown,
+    Found { id: LeaderId },
+}
+
+pub struct Key(String);
+
+impl From<String> for Key {
+    fn from(value: String) -> Self {
+        Self(value)
+    }
+}
+impl From<&str> for Key {
+    fn from(value: &str) -> Self {
+        Self(value.to_string())
+    }
+}
+
+impl From<Key> for String {
+    fn from(value: Key) -> Self {
+        value.0
+    }
+}
+impl From<&Key> for String {
+    fn from(value: &Key) -> Self {
+        value.0.clone()
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct LeaderId(uuid::Uuid);
+impl LeaderId {
+    pub(crate) fn new() -> Self {
+        Self(uuid::Uuid::now_v7())
+    }
+}
+
+impl From<LeaderId> for uuid::Uuid {
+    fn from(value: LeaderId) -> Self {
+        value.0
+    }
+}
+impl From<&LeaderId> for uuid::Uuid {
+    fn from(value: &LeaderId) -> Self {
+        value.0
+    }
+}
+
+impl From<uuid::Uuid> for LeaderId {
+    fn from(value: uuid::Uuid) -> Self {
+        Self(value)
+    }
+}
+
+impl LeaderId {
+    pub const fn as_bytes(&self) -> &[u8] {
+        self.0.as_bytes()
+    }
+}

crates/noleader/src/backend/nats.rs

@@ -0,0 +1,131 @@
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use anyhow::Context;
+use async_nats::jetstream::{self, kv};
+
+use crate::backend::{BackendEdge, Key, LeaderId, LeaderValue};
+
+pub struct NatsBackend {
+    bucket: String,
+    client: jetstream::Context,
+
+    revision: AtomicU64,
+}
+
+impl NatsBackend {
+    pub fn new(client: async_nats::Client, bucket: &str) -> Self {
+        Self {
+            bucket: bucket.into(),
+            client: jetstream::new(client),
+            revision: AtomicU64::new(0),
+        }
+    }
+
+    pub async fn create_bucket(&self) -> anyhow::Result<()> {
+        if (self.client.get_key_value(&self.bucket).await).is_ok() {
+            return Ok(());
+        }
+
+        if let Err(e) = self
+            .client
+            .create_key_value(kv::Config {
+                bucket: self.bucket.clone(),
+                description: "leadership bucket for noleader".into(),
+                limit_markers: Some(std::time::Duration::from_secs(60)),
+                max_age: std::time::Duration::from_secs(60),
+                ..Default::default()
+            })
+            .await
+        {
+            tracing::info!(
+                "bucket creation failed, it might have just been a conflict, testing again: {e}"
+            );
+
+            if (self.client.get_key_value(&self.bucket).await).is_ok() {
+                return Ok(());
+            }
+
+            anyhow::bail!("failed to create bucket: {}", e)
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl BackendEdge for NatsBackend {
+    async fn setup(&self) -> anyhow::Result<()> {
+        self.create_bucket().await?;
+
+        Ok(())
+    }
+    async fn get(&self, key: &Key) -> anyhow::Result<LeaderValue> {
+        let bucket = self.client.get_key_value(&self.bucket).await?;
+
+        let Some(val) = bucket.get(key).await? else {
+            anyhow::bail!("key doesn't exists, we've lost leadership status")
+        };
+
+        let Ok(id) = uuid::Uuid::from_slice(&val) else {
+            return Ok(LeaderValue::Unknown);
+        };
+
+        Ok(LeaderValue::Found { id: id.into() })
+    }
+    async fn update(&self, key: &Key, val: &LeaderId) -> anyhow::Result<()> {
+        let bucket = self
+            .client
+            .get_key_value(&self.bucket)
+            .await
+            .context("get bucket")?;
+
+        match bucket
+            .update(
+                &key.0,
+                bytes::Bytes::copy_from_slice(val.as_bytes()),
+                self.revision.load(Ordering::Relaxed),
+            )
+            .await
+        {
+            Ok(rev) => {
+                self.revision.store(rev, Ordering::Relaxed);
+            }
+            Err(e) => match e.kind() {
+                kv::UpdateErrorKind::WrongLastRevision => {
+                    tracing::trace!("creating nats entry");
+                    match bucket
+                        .create_with_ttl(
+                            &key.0,
+                            bytes::Bytes::copy_from_slice(val.as_bytes()),
+                            std::time::Duration::from_secs(60),
+                        )
+                        .await
+                    {
+                        Ok(rev) => {
+                            self.revision.store(rev, Ordering::Relaxed);
+                        }
+                        Err(e) => match e.kind() {
+                            kv::CreateErrorKind::AlreadyExists => {
+                                anyhow::bail!("another candidate has leadership status")
+                            }
+                            _ => {
+                                anyhow::bail!("{}", e);
+                            }
+                        },
+                    }
+                }
+                _ => {
+                    anyhow::bail!("failed to create bucket: {e}")
+                }
+            },
+        }
+
+        Ok(())
+    }
+
+    async fn release(&self, _key: &Key, _val: &LeaderId) -> anyhow::Result<()> {
+        // TODO: implement release for nats
+
+        Ok(())
+    }
+}

crates/noleader/src/backend/postgres.rs

@@ -0,0 +1,224 @@
+use std::{
+    sync::atomic::{AtomicU64, Ordering},
+    time::Duration,
+};
+
+use anyhow::Context;
+use sqlx::{PgPool, postgres::PgPoolOptions};
+use tokio::sync::OnceCell;
+
+use crate::backend::{BackendEdge, Key, LeaderId, LeaderValue};
+
+pub struct PostgresBackend {
+    database_url: String,
+    revision: AtomicU64,
+    pool: OnceCell<PgPool>,
+    migrated: OnceCell<()>,
+}
+
+impl PostgresBackend {
+    pub fn new(database_url: &str) -> Self {
+        Self {
+            database_url: database_url.into(),
+            revision: AtomicU64::new(0),
+            pool: OnceCell::new(),
+            migrated: OnceCell::new(),
+        }
+    }
+
+    pub fn new_with_pool(database_url: &str, pool: PgPool) -> Self {
+        Self {
+            database_url: database_url.into(),
+            revision: AtomicU64::new(0),
+            pool: OnceCell::new_with(Some(pool)),
+            migrated: OnceCell::new(),
+        }
+    }
+
+    async fn db(&self) -> anyhow::Result<PgPool> {
+        let pool = self
+            .pool
+            .get_or_try_init(|| async move {
+                PgPoolOptions::new()
+                    .max_connections(1)
+                    .min_connections(0)
+                    .idle_timeout(Some(Duration::from_secs(5)))
+                    .connect_lazy(&self.database_url)
+                    .context("connect postgres noleader")
+            })
+            .await?;
+
+        Ok(pool.clone())
+    }
+
+    async fn migrate(&self) -> anyhow::Result<()> {
+        self.migrated
+            .get_or_try_init(|| async move {
+                let db = self.db().await?;
+
+                let mut migrate = sqlx::migrate!("./migrations/postgres/");
+
+                migrate
+                    .set_locking(false)
+                    .dangerous_set_table_name("_sqlx_noleader_migrations")
+                    .run(&db)
+                    .await
+                    .context("migrate noleader")?;
+
+                Ok::<_, anyhow::Error>(())
+            })
+            .await?;
+
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl BackendEdge for PostgresBackend {
+    async fn setup(&self) -> anyhow::Result<()> {
+        self.migrate().await?;
+        Ok(())
+    }
+
+    async fn get(&self, key: &Key) -> anyhow::Result<LeaderValue> {
+        let rec: Option<GetResult> = sqlx::query_as(
+            "
+            SELECT value, revision
+            FROM noleader_leaders
+            WHERE
+                  key = $1
+              AND heartbeat >= now() - interval '60 seconds'
+            LIMIT 1;
+            ",
+        )
+        .bind(&key.0)
+        .fetch_optional(&self.db().await?)
+        .await
+        .context("get noleader key")?;
+
+        let Some(val) = rec else {
+            anyhow::bail!("key doesn't exist, we've lost leadership status")
+        };
+
+        // Update our local revision to match what's in the database
+        self.revision.store(val.revision as u64, Ordering::Relaxed);
+
+        let Ok(id) = uuid::Uuid::parse_str(&val.value) else {
+            tracing::warn!("value is not a valid uuid: {}", val.value);
+            return Ok(LeaderValue::Unknown);
+        };
+
+        Ok(LeaderValue::Found { id: id.into() })
+    }
+
+    async fn update(&self, key: &Key, val: &LeaderId) -> anyhow::Result<()> {
+        let current_rev = self.revision.load(Ordering::Relaxed);
+        let new_rev = current_rev + 1;
+
+        let res: Result<Option<UpdateResult>, sqlx::Error> = sqlx::query_as(
+            r#"
+            INSERT INTO noleader_leaders (key, value, revision, heartbeat)
+            VALUES ($1, $2, $3, now())
+            ON CONFLICT (key)
+            DO UPDATE SET
+                value = EXCLUDED.value,
+                revision = EXCLUDED.revision,
+                heartbeat = now()
+            WHERE 
+                (
+                    -- Normal case: revision matches (we're the current leader updating)
+                    noleader_leaders.revision = $4
+                    OR
+                    -- Override case: heartbeat is old (stale leader)
+                    noleader_leaders.heartbeat < now() - INTERVAL '60 seconds'
+                )
+            RETURNING value, revision
+            "#,
+        )
+        .bind(&key.0)
+        .bind(val.0.to_string())
+        .bind(new_rev as i64) // new revision
+        .bind(current_rev as i64) // expected current revision
+        .fetch_optional(&self.db().await?)
+        .await;
+
+        let res = match res {
+            Ok(res) => res,
+            Err(e) => match &e {
+                sqlx::Error::Database(database_error) => {
+                    if database_error.is_unique_violation() {
+                        anyhow::bail!("update conflict: another leader holds lock")
+                    } else {
+                        anyhow::bail!(e);
+                    }
+                }
+                _ => {
+                    anyhow::bail!(e);
+                }
+            },
+        };
+
+        match res {
+            Some(rec) => {
+                if rec.value == val.0.to_string() && rec.revision == new_rev as i64 {
+                    tracing::debug!(
+                        val = val.0.to_string(),
+                        revision = rec.revision,
+                        "successfully updated leader"
+                    );
+
+                    // Only update our local revision if the update succeeded with our expected value
+                    self.revision.store(rec.revision as u64, Ordering::Relaxed);
+                } else {
+                    anyhow::bail!(
+                        "update conflict: expected value={}, revision={}, got value={}, revision={}",
+                        val.0.to_string(),
+                        new_rev,
+                        rec.value,
+                        rec.revision
+                    );
+                }
+            }
+            None => {
+                anyhow::bail!(
+                    "update rejected: another leader is holding the lock or revision mismatch"
+                )
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn release(&self, key: &Key, val: &LeaderId) -> anyhow::Result<()> {
+        let rev = self.revision.load(Ordering::Relaxed);
+        sqlx::query(
+            "
+                DELETE FROM noleader_leaders
+                WHERE
+                        key = $1
+                    AND value = $2
+                    AND revision = $3
+            ",
+        )
+        .bind(&key.0)
+        .bind(val.0.to_string())
+        .bind(rev as i64) // new revision
+        .execute(&self.db().await?)
+        .await
+        .context("failed to release lock, it will expire naturally")?;
+
+        Ok(())
+    }
+}
+
+#[derive(sqlx::FromRow)]
+struct GetResult {
+    value: String,
+    revision: i64,
+}
+
+#[derive(sqlx::FromRow)]
+struct UpdateResult {
+    value: String,
+    revision: i64,
+}

crates/noleader/src/lib.rs

@@ -1,54 +1,168 @@
 use std::{
     future::Future,
     sync::{
-        atomic::{AtomicBool, Ordering},
         Arc,
+        atomic::{AtomicBool, Ordering},
     },
     time::Duration,
 };
 
 use anyhow::Context;
-use async_nats::jetstream::kv;
 use rand::Rng;
 use tokio::sync::RwLock;
 use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
+use crate::backend::{Backend, Key, LeaderId};
+
 #[derive(Clone)]
 pub struct Leader {
     shutting_down: Arc<AtomicBool>,
     is_leader: Arc<AtomicBool>,
     inner: Arc<RwLock<InnerLeader>>,
+
+    cancellation: CancellationToken,
 }
 const DEFAULT_INTERVAL: Duration = std::time::Duration::from_secs(10);
 
+mod backend;
+
 impl Leader {
-    pub fn new(bucket: &str, key: &str, client: async_nats::Client) -> Self {
+    pub fn new(key: &str, backend: Backend) -> Self {
         Self {
             shutting_down: Arc::new(AtomicBool::new(false)),
             is_leader: Arc::new(AtomicBool::new(false)),
-            inner: Arc::new(RwLock::new(InnerLeader::new(bucket, key, client))),
+            inner: Arc::new(RwLock::new(InnerLeader::new(backend, key))),
+            cancellation: CancellationToken::new(),
         }
     }
 
+    #[cfg(feature = "nats")]
+    pub fn new_nats(key: &str, bucket: &str, client: async_nats::Client) -> Self {
+        Self::new(key, Backend::nats(client, bucket))
+    }
+
+    #[cfg(feature = "postgres")]
+    pub fn new_postgres(key: &str, database_url: &str) -> Self {
+        Self::new(key, Backend::postgres(database_url))
+    }
+
+    #[cfg(feature = "postgres")]
+    pub fn new_postgres_pool(key: &str, pool: sqlx::PgPool) -> Self {
+        Self::new(key, Backend::postgres_with_pool(pool))
+    }
+
+    pub fn with_cancellation(&mut self, cancellation: CancellationToken) -> &mut Self {
+        self.cancellation = cancellation;
+        self
+    }
+
+    pub fn with_cancel_task<T>(&mut self, f: T) -> &mut Self
+    where
+        T: Future<Output = ()> + Send + 'static,
+    {
+        let cancel = self.cancellation.clone();
+
+        tokio::spawn(async move {
+            f.await;
+
+            cancel.cancel();
+        });
+
+        self
+    }
+
+    pub async fn acquire_and_run<F, Fut>(&self, f: F) -> anyhow::Result<()>
+    where
+        F: Fn(CancellationToken) -> Fut,
+        Fut: Future<Output = anyhow::Result<()>> + Send + 'static,
+    {
+        let parent_token = self.cancellation.clone();
+        let s = self.clone();
+
+        let server_token = parent_token.child_token();
+
+        // Start the server election process in another task, this is because start is blocking
+        let handle = tokio::spawn({
+            async move {
+                match s.start().await {
+                    Ok(_) => {}
+                    Err(e) => tracing::error!("leader election process failed: {}", e),
+                }
+
+                tracing::info!("shutting down noleader");
+
+                parent_token.cancel();
+            }
+        });
+
+        // Do the work if we're leader
+        let res = self
+            .do_while_leader_inner(server_token.child_token(), f)
+            .await;
+
+        // Stop the server election process if our provided functions returns an error;
+        server_token.cancel();
+        // Close down the task as well, it should already be stopped, but this forces the task to close
+        handle.abort();
+
+        {
+            self.inner.write().await.cleanup().await?;
+        }
+
+        res?;
+
+        Ok(())
+    }
+
     pub async fn do_while_leader<F, Fut>(&self, f: F) -> anyhow::Result<()>
+    where
+        F: Fn(CancellationToken) -> Fut,
+        Fut: Future<Output = anyhow::Result<()>> + Send + 'static,
+    {
+        self.do_while_leader_inner(CancellationToken::new(), f)
+            .await
+    }
+
+    async fn do_while_leader_inner<F, Fut>(
+        &self,
+        cancellation_token: CancellationToken,
+        f: F,
+    ) -> anyhow::Result<()>
     where
         F: Fn(CancellationToken) -> Fut,
         Fut: Future<Output = anyhow::Result<()>> + Send + 'static,
     {
         loop {
+            if cancellation_token.is_cancelled() {
+                return Ok(());
+            }
+
+            let cancellation_token = cancellation_token.child_token();
+
             let is_leader = self.is_leader.clone();
             if !is_leader.load(Ordering::Relaxed) {
-                tokio::time::sleep(Duration::from_secs(1)).await;
+                tokio::select! {
+                    _ = tokio::time::sleep(Duration::from_secs(1)) => {}
+                    _ = cancellation_token.cancelled() => {
+                        return Ok(());
+                    }
+                }
+
                 continue;
             }
 
-            let cancellation_token = CancellationToken::new();
             let child_token = cancellation_token.child_token();
 
             let guard = tokio::spawn(async move {
                 loop {
-                    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+                    tokio::select! {
+                        _ = tokio::time::sleep(std::time::Duration::from_millis(500)) => {}
+                        _ = cancellation_token.cancelled() => {
+                            return;
+                        }
+
+                    }
 
                     if !is_leader.load(Ordering::Relaxed) {
                         cancellation_token.cancel();
@@ -57,6 +171,7 @@ impl Leader {
             });
 
             let res = f(child_token).await;
+
             guard.abort();
             res?;
         }
@@ -64,35 +179,30 @@ impl Leader {
 
     pub async fn leader_id(&self) -> Uuid {
         let inner = self.inner.read().await;
-        inner.id
+        inner.leader_id.clone().into()
     }
 
-    pub async fn create_bucket(&self) -> anyhow::Result<()> {
+    pub async fn start(&self) -> anyhow::Result<()> {
-        let mut inner = self.inner.write().await;
-        tracing::info!("creating bucket leadership bucket");
-
-        inner.create_bucket().await?;
-
-        Ok(())
-    }
-
-    pub async fn start(&self, cancellation_token: CancellationToken) -> anyhow::Result<()> {
         let mut attempts = 1;
 
+        {
+            self.inner.write().await.backend.setup().await?;
+        }
+
         // Initial attempt
         let _ = self.try_become_leader().await;
 
         loop {
             let wait_factor = {
                 let mut rng = rand::rng();
-                rng.random_range(0.001..1.000)
+                rng.random_range(0.50..1.00)
             };
 
             let sleep_fut = tokio::time::sleep((DEFAULT_INTERVAL * attempts).mul_f64(wait_factor));
 
             tokio::select! {
                 _ = sleep_fut => {},
-                _ = cancellation_token.cancelled() => {
+                _ = self.cancellation.cancelled() => {
                     self.shutting_down.store(true, std::sync::atomic::Ordering::Relaxed); // Ordering can be relaxed, because our operation is an atomic update
                     return Ok(())
                 }
@@ -100,8 +210,7 @@ impl Leader {
 
             match self.try_become_leader().await {
                 Ok(_) => {
-                    self.is_leader
+                    self.is_leader.store(true, Ordering::Relaxed);
-                        .store(true, std::sync::atomic::Ordering::Relaxed);
                     attempts = 1;
                 }
                 Err(e) => {
@@ -150,13 +259,10 @@ pub enum Status {
 struct InnerLeader {
     state: LeaderState,
 
-    bucket: String,
+    backend: Backend,
-    key: String,
 
-    id: uuid::Uuid,
+    key: Key,
-    revision: u64,
+    leader_id: LeaderId,
-
-    client: async_nats::jetstream::Context,
 }
 
 #[derive(Default, Clone)]
@@ -168,49 +274,17 @@ enum LeaderState {
 }
 
 impl InnerLeader {
-    pub fn new(bucket: &str, key: &str, client: async_nats::Client) -> Self {
+    pub fn new(backend: Backend, key: impl Into<Key>) -> Self {
         Self {
-            bucket: bucket.into(),
+            backend,
+            leader_id: LeaderId::new(),
+
             key: key.into(),
 
-            id: uuid::Uuid::new_v4(),
-            revision: u64::MIN,
-
             state: LeaderState::Unknown,
-            client: async_nats::jetstream::new(client),
         }
     }
 
-    pub async fn create_bucket(&mut self) -> anyhow::Result<()> {
-        if (self.client.get_key_value(&self.bucket).await).is_ok() {
-            return Ok(());
-        }
-
-        if let Err(e) = self
-            .client
-            .create_key_value(kv::Config {
-                bucket: self.bucket.clone(),
-                description: "leadership bucket for noleader".into(),
-                limit_markers: Some(std::time::Duration::from_secs(60)),
-                max_age: std::time::Duration::from_secs(60),
-                ..Default::default()
-            })
-            .await
-        {
-            tracing::info!(
-                "bucket creation failed, it might have just been a conflict, testing again: {e}"
-            );
-
-            if (self.client.get_key_value(&self.bucket).await).is_ok() {
-                return Ok(());
-            }
-
-            anyhow::bail!("failed to create bucket: {}", e)
-        }
-
-        Ok(())
-    }
-
     /// start, will run a blocking operation for becoming the next leader.
     pub async fn start(&mut self) -> anyhow::Result<()> {
         // Attempt to grab leadership,
@@ -248,61 +322,42 @@ impl InnerLeader {
         Ok(())
     }
 
-    async fn update_leadership(&mut self) -> anyhow::Result<()> {
+    pub async fn cleanup(&self) -> anyhow::Result<()> {
-        let bucket = self.client.get_key_value(&self.bucket).await?;
+        self.backend
+            .release(&self.key, &self.leader_id)
+            .await
+            .context("cleanup")?;
 
-        let Some(val) = bucket.get(&self.key).await? else {
+        Ok(())
-            anyhow::bail!("key doesn't exists, we've lost leadership status")
-        };
-
-        let Ok(id) = uuid::Uuid::from_slice(&val) else {
-            anyhow::bail!("value has changed, it is no longer a uuid, dropping leadership status");
-        };
-
-        if id != self.id {
-            anyhow::bail!("leadership has changed")
     }
 
-        let rev = bucket
+    async fn update_leadership(&mut self) -> anyhow::Result<()> {
-            .update(
+        let val = self
-                &self.key,
+            .backend
-                bytes::Bytes::copy_from_slice(self.id.as_bytes()),
+            .get(&self.key)
-                self.revision,
+            .await
-            )
+            .context("could not find key, we've lost leadership status")?;
-            .await?;
 
-        self.revision = rev;
+        match val {
+            backend::LeaderValue::Unknown => anyhow::bail!("leadership is unknown"),
+            backend::LeaderValue::Found { id } if id != self.leader_id => {
+                anyhow::bail!("leadership has changed")
+            }
+            backend::LeaderValue::Found { .. } => self
+                .backend
+                .update(&self.key, &self.leader_id)
+                .await
+                .context("update leadership lock")?,
+        }
 
         Ok(())
     }
 
     async fn try_for_leadership(&mut self) -> anyhow::Result<()> {
-        let bucket = self
+        self.backend
-            .client
+            .update(&self.key, &self.leader_id)
-            .get_key_value(&self.bucket)
             .await
-            .context("failed to get bucket")?;
+            .context("try for leadership")?;
-
-        let rev = match bucket
-            .create_with_ttl(
-                &self.key,
-                bytes::Bytes::copy_from_slice(self.id.as_bytes()),
-                std::time::Duration::from_secs(60),
-            )
-            .await
-        {
-            Ok(rev) => rev,
-            Err(e) => match e.kind() {
-                kv::CreateErrorKind::AlreadyExists => {
-                    anyhow::bail!("another candidate has leadership status")
-                }
-                _ => {
-                    anyhow::bail!("{}", e);
-                }
-            },
-        };
-
-        self.revision = rev;
 
         tokio::time::sleep(DEFAULT_INTERVAL).await;
 
@@ -310,7 +365,7 @@ impl InnerLeader {
 
         let leadership_state = self.leadership_status().await?;
 
-        if !leadership_state.is_leader(&self.id) {
+        if !leadership_state.is_leader(&self.leader_id) {
             anyhow::bail!("failed to become leader, there is likely some churn going on");
         }
 
@@ -321,25 +376,16 @@ impl InnerLeader {
     }
 
     async fn leadership_status(&mut self) -> anyhow::Result<LeadershipState> {
-        let bucket = self.client.get_key_value(&self.bucket).await?;
+        let val = self
-
+            .backend
-        let val = bucket.get(&self.key).await?;
+            .get(&self.key)
+            .await
+            .inspect_err(|e| tracing::warn!("failed to query for leadership: {}", e))
+            .ok();
 
         Ok(match val {
-            Some(content) => {
+            Some(backend::LeaderValue::Found { id }) => LeadershipState::Allocated { id },
-                let id = match uuid::Uuid::from_slice(&content) {
+            Some(backend::LeaderValue::Unknown) => LeadershipState::NotFound,
-                    Ok(u) => u,
-                    Err(e) => {
-                        tracing::warn!(
-                            "leadership state is not a valid UUID, ignoring the value: {}",
-                            e
-                        );
-                        return Ok(LeadershipState::NotFound);
-                    }
-                };
-
-                LeadershipState::Allocated { id }
-            }
             None => LeadershipState::NotFound,
         })
     }
@@ -347,11 +393,11 @@ impl InnerLeader {
 
 enum LeadershipState {
     NotFound,
-    Allocated { id: uuid::Uuid },
+    Allocated { id: LeaderId },
 }
 
 impl LeadershipState {
-    pub fn is_leader(&self, leader_id: &Uuid) -> bool {
+    pub fn is_leader(&self, leader_id: &LeaderId) -> bool {
         match self {
             LeadershipState::Allocated { id } => id == leader_id,
             _ => false,

mise.toml

@@ -0,0 +1,20 @@
+[env]
+_.file = ".env"
+
+
+[tasks.test]
+alias = ["t"]
+run = "cargo nextest run"
+
+[tasks.example]
+alias = ["e"]
+run = "cargo run --example"
+
+[tasks."local:up"]
+run = "docker compose -f ./templates/docker/docker-compose.yml up -d"
+
+[tasks."local:down"]
+run = "docker compose -f ./templates/docker/docker-compose.yml down -v"
+
+[tasks."db:prepare"]
+run = "cargo sqlx prepare --workspace"

templates/docker/docker-compose.yml

@@ -7,3 +7,18 @@ services:
       - "4222:4222"   # Client connections
       - "8222:8222"   # HTTP monitoring
       - "6222:6222"   # Clustering
+
+  postgres:
+    image: postgres:17-alpine
+    environment:
+      POSTGRES_USER: devuser
+      POSTGRES_PASSWORD: devpassword
+      POSTGRES_DB: dev
+    shm_size: 128mb
+    ports:
+      - "5432:5432"
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U devuser -d dev"]
+      interval: 5s
+      timeout: 5s
+      retries: 5