Repository: stacks-network/stacks-core (language: python)

Issue:
feat: make the timeout for event observers configurable
### Description
Adds an optional `timeout_ms` to the configuration for event observers, allowing users to set a unique timeout for each observer.

### Checklist

- [x] Test coverage for new or modified code paths
- [x] Changelog is updated
- [ ] Required documentation changes (e.g., `docs/rpc/openapi.yaml` and `rpc-endpoints.md` for v2 endpoints, `event-dispatcher.md` for new events)
- [ ] New clarity functions have corresponding PR in `clarity-benchmarking` repo
- [ ] New integration test(s) added to `bitcoin-tests.yml`



Discussion / hints:
@obycode Would it be better to err on the side of flexibility and change the default to something larger that _wouldn't_ likely cause issues syncing a large block to a low-powered observer, like an API with minimal resources?
Instead of setting the default to 1 second, it might be better to set it to 60 seconds by default and let others rein it in if needed. Thoughts on this?
> @obycode Would it be better to err on the side of flexibility and change the default to something larger that _wouldn't_ likely cause issues syncing a large block to a low-powered observer, like an API with minimal resources? Instead of setting the default to 1 second, it might be better to set it to 60 seconds by default and let others rein it in if needed. Thoughts on this?

Yeah, that could make sense. I chose 1s as the default since that what it was hard-coded to before this change, so this change would have no effect.

The other problem with defaulting to 60s is that that could be surprising, since it halts progression of a node, a 60s timeout before seeing any logs is a lot, given that blocks will be coming in on the order of seconds.
for what it's worth, confirmed that this change does allow genesis sync with the API (moved past block 0)
> The other problem with defaulting to 60s is that that could be surprising, since it halts progression of a node, a 60s timeout before seeing any logs is a lot, given that blocks will be coming in on the order of seconds.

I agree, but I think that says more about the logging from when a timeout is reached rather than the default setting itself. If the log when an observer timeout is specific, like "timeout reached (60s) trying to send block 12345 to observer my.domain.com:3999", it wouldn't be very confusing to debug.

Not suggesting this is needed now, but just thinking it would pose less consternation to set a wide boundary now and come back to it later with improved logging, rather than set a narrow boundary now and cause many operators to update all of their configs since 1s is likely too low for many people, and JW has already seen payloads take close to 30 seconds.

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5ccb9b5cac..50d149b730 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,6 +18,7 @@ and this project adheres to the versioning scheme outlined in the [README.md](RE
     - `get-block-info?` removed
 - Added `/v3/signer/{signer_pubkey}/{reward_cycle}` endpoint
 - Added optional `timeout_ms` to `events_observer` configuration
+- Added support for re-sending events to event observers across restarts
 
 ## [2.5.0.0.7]
 
diff --git a/Cargo.lock b/Cargo.lock
index dc27c931cc..227cd9d768 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -190,6 +190,16 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "assert-json-diff"
+version = "2.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47e4f2b81832e72834d7518d8487a0396a28cc408186a2e8854c0f98011faf12"
+dependencies = [
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "async-attributes"
 version = "1.1.2"
@@ -626,7 +636,7 @@ checksum = "98cc8fbded0c607b7ba9dd60cd98df59af97e84d24e49c8557331cfc26d301ce"
 name = "clarity"
 version = "0.0.1"
 dependencies = [
- "assert-json-diff",
+ "assert-json-diff 1.1.0",
  "hashbrown",
  "integer-sqrt",
  "lazy_static",
@@ -652,6 +662,16 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "acbf1af155f9b9ef647e42cdc158db4b64a1b61f743629225fde6f3e0be2a7c7"
 
+[[package]]
+name = "colored"
+version = "2.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cbf2150cce219b664a8a70df7a1f933836724b503f8a413af9365b4dcc4d90b8"
+dependencies = [
+ "lazy_static",
+ "windows-sys 0.48.0",
+]
+
 [[package]]
 name = "concurrent-queue"
 version = "2.4.0"
@@ -1347,7 +1367,26 @@ dependencies = [
  "futures-core",
  "futures-sink",
  "futures-util",
- "http",
+ "http 0.2.11",
+ "indexmap",
+ "slab",
+ "tokio",
+ "tokio-util",
+ "tracing",
+]
+
+[[package]]
+name = "h2"
+version = "0.4.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "524e8ac6999421f49a846c2d4411f337e53497d8ec55d67753beffa43c5d9205"
+dependencies = [
+ "atomic-waker",
+ "bytes",
+ "fnv",
+ "futures-core",
+ "futures-sink",
+ "http 1.1.0",
  "indexmap",
  "slab",
  "tokio",
@@ -1390,7 +1429,7 @@ dependencies = [
  "base64 0.21.7",
  "bytes",
  "headers-core",
- "http",
+ "http 0.2.11",
  "httpdate",
  "mime",
  "sha1 0.10.6",
@@ -1402,7 +1441,7 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e7f66481bfee273957b1f20485a4ff3362987f85b2c236580d81b4eb7a326429"
 dependencies = [
- "http",
+ "http 0.2.11",
 ]
 
 [[package]]
@@ -1457,6 +1496,17 @@ dependencies = [
  "itoa",
 ]
 
+[[package]]
+name = "http"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "21b9ddb458710bc376481b842f5da65cdf31522de232c1ca8146abce2a358258"
+dependencies = [
+ "bytes",
+ "fnv",
+ "itoa",
+]
+
 [[package]]
 name = "http-body"
 version = "0.4.6"
@@ -1464,7 +1514,30 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2"
 dependencies = [
  "bytes",
- "http",
+ "http 0.2.11",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "http-body"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184"
+dependencies = [
+ "bytes",
+ "http 1.1.0",
+]
+
+[[package]]
+name = "http-body-util"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "793429d76616a256bcb62c2a2ec2bed781c8307e797e2598c50010f2bee2544f"
+dependencies = [
+ "bytes",
+ "futures-util",
+ "http 1.1.0",
+ "http-body 1.0.1",
  "pin-project-lite",
 ]
 
@@ -1512,9 +1585,9 @@ dependencies = [
  "futures-channel",
  "futures-core",
  "futures-util",
- "h2",
- "http",
- "http-body",
+ "h2 0.3.24",
+ "http 0.2.11",
+ "http-body 0.4.6",
  "httparse",
  "httpdate",
  "itoa",
@@ -1526,6 +1599,26 @@ dependencies = [
  "want",
 ]
 
+[[package]]
+name = "hyper"
+version = "1.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "50dfd22e0e76d0f662d429a5f80fcaf3855009297eab6a0a9f8543834744ba05"
+dependencies = [
+ "bytes",
+ "futures-channel",
+ "futures-util",
+ "h2 0.4.6",
+ "http 1.1.0",
+ "http-body 1.0.1",
+ "httparse",
+ "httpdate",
+ "itoa",
+ "pin-project-lite",
+ "smallvec",
+ "tokio",
+]
+
 [[package]]
 name = "hyper-rustls"
 version = "0.24.2"
@@ -1533,13 +1626,28 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ec3efd23720e2049821a693cbc7e65ea87c72f1c58ff2f9522ff332b1491e590"
 dependencies = [
  "futures-util",
- "http",
- "hyper",
+ "http 0.2.11",
+ "hyper 0.14.28",
  "rustls",
  "tokio",
  "tokio-rustls",
 ]
 
+[[package]]
+name = "hyper-util"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cde7055719c54e36e95e8719f95883f22072a48ede39db7fc17a4e1d5281e9b9"
+dependencies = [
+ "bytes",
+ "futures-util",
+ "http 1.1.0",
+ "http-body 1.0.1",
+ "hyper 1.4.1",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "iana-time-zone"
 version = "0.1.60"
@@ -1725,7 +1833,7 @@ checksum = "85c833ca1e66078851dba29046874e38f08b2c883700aa29a03ddd3b23814ee8"
 dependencies = [
  "bitflags 2.4.2",
  "libc",
- "redox_syscall",
+ "redox_syscall 0.4.1",
 ]
 
 [[package]]
@@ -1792,6 +1900,16 @@ version = "0.4.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "01cda141df6706de531b6c46c3a33ecca755538219bd484262fa09410c13539c"
 
+[[package]]
+name = "lock_api"
+version = "0.4.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "07af8b9cdd281b7915f413fa73f29ebd5d55d0d3f0155584dade1ff18cea1b17"
+dependencies = [
+ "autocfg",
+ "scopeguard",
+]
+
 [[package]]
 name = "log"
 version = "0.4.20"
@@ -1892,6 +2010,30 @@ dependencies = [
  "ws2_32-sys",
 ]
 
+[[package]]
+name = "mockito"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09b34bd91b9e5c5b06338d392463e1318d683cf82ec3d3af4014609be6e2108d"
+dependencies = [
+ "assert-json-diff 2.0.2",
+ "bytes",
+ "colored",
+ "futures-util",
+ "http 1.1.0",
+ "http-body 1.0.1",
+ "http-body-util",
+ "hyper 1.4.1",
+ "hyper-util",
+ "log",
+ "rand 0.8.5",
+ "regex",
+ "serde_json",
+ "serde_urlencoded",
+ "similar",
+ "tokio",
+]
+
 [[package]]
 name = "multer"
 version = "2.1.0"
@@ -1901,7 +2043,7 @@ dependencies = [
  "bytes",
  "encoding_rs",
  "futures-util",
- "http",
+ "http 0.2.11",
  "httparse",
  "log",
  "memchr",
@@ -2023,6 +2165,29 @@ version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bb813b8af86854136c6922af0598d719255ecb2179515e6e7730d468f05c9cae"
 
+[[package]]
+name = "parking_lot"
+version = "0.12.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f1bf18183cf54e8d6059647fc3063646a1801cf30896933ec2311622cc4b9a27"
+dependencies = [
+ "lock_api",
+ "parking_lot_core",
+]
+
+[[package]]
+name = "parking_lot_core"
+version = "0.9.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e401f977ab385c9e4e3ab30627d6f26d00e2c73eef317493c4ec6d468726cf8"
+dependencies = [
+ "cfg-if 1.0.0",
+ "libc",
+ "redox_syscall 0.5.7",
+ "smallvec",
+ "windows-targets 0.52.0",
+]
+
 [[package]]
 name = "percent-encoding"
 version = "2.3.1"
@@ -2388,6 +2553,15 @@ dependencies = [
  "bitflags 1.3.2",
 ]
 
+[[package]]
+name = "redox_syscall"
+version = "0.5.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9b6dfecf2c74bce2466cabf93f6664d6998a69eb21e39f4207930065b27b771f"
+dependencies = [
+ "bitflags 2.4.2",
+]
+
 [[package]]
 name = "redox_users"
 version = "0.4.4"
@@ -2461,10 +2635,10 @@ dependencies = [
  "encoding_rs",
  "futures-core",
  "futures-util",
- "h2",
- "http",
- "http-body",
- "hyper",
+ "h2 0.3.24",
+ "http 0.2.11",
+ "http-body 0.4.6",
+ "hyper 0.14.28",
  "hyper-rustls",
  "ipnet",
  "js-sys",
@@ -2739,6 +2913,12 @@ version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294"
 
+[[package]]
+name = "scopeguard"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
+
 [[package]]
 name = "sct"
 version = "0.7.1"
@@ -2969,6 +3149,12 @@ dependencies = [
  "rand_core 0.6.4",
 ]
 
+[[package]]
+name = "similar"
+version = "2.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1de1d4f81173b03af4c0cbed3c898f6bff5b870e4a7f5d6f4057d62a7a4b686e"
+
 [[package]]
 name = "siphasher"
 version = "0.3.11"
@@ -3080,7 +3266,7 @@ dependencies = [
 name = "stacks-common"
 version = "0.0.1"
 dependencies = [
- "assert-json-diff",
+ "assert-json-diff 1.1.0",
  "chrono",
  "curve25519-dalek 2.0.0",
  "ed25519-dalek",
@@ -3124,6 +3310,7 @@ dependencies = [
  "lazy_static",
  "libc",
  "libsigner",
+ "mockito",
  "mutants",
  "pico-args",
  "rand 0.8.5",
@@ -3140,6 +3327,7 @@ dependencies = [
  "stacks-signer",
  "stackslib",
  "stx-genesis",
+ "tempfile",
  "tikv-jemallocator",
  "tiny_http",
  "tokio",
@@ -3190,7 +3378,7 @@ dependencies = [
 name = "stackslib"
 version = "0.0.1"
 dependencies = [
- "assert-json-diff",
+ "assert-json-diff 1.1.0",
  "chrono",
  "clarity",
  "criterion",
@@ -3369,6 +3557,19 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "tempfile"
+version = "3.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8fcd239983515c23a32fb82099f97d0b11b8c72f654ed659363a95c3dad7a53"
+dependencies = [
+ "cfg-if 1.0.0",
+ "fastrand 2.0.1",
+ "once_cell",
+ "rustix 0.38.31",
+ "windows-sys 0.52.0",
+]
+
 [[package]]
 name = "term"
 version = "0.7.0"
@@ -3558,6 +3759,7 @@ dependencies = [
  "libc",
  "mio 0.8.10",
  "num_cpus",
+ "parking_lot",
  "pin-project-lite",
  "socket2 0.5.5",
  "windows-sys 0.48.0",
@@ -3702,7 +3904,7 @@ dependencies = [
  "byteorder",
  "bytes",
  "data-encoding",
- "http",
+ "http 0.2.11",
  "httparse",
  "log",
  "rand 0.8.5",
@@ -3865,8 +4067,8 @@ dependencies = [
  "futures-channel",
  "futures-util",
  "headers",
- "http",
- "hyper",
+ "http 0.2.11",
+ "hyper 0.14.28",
  "log",
  "mime",
  "mime_guess",


Test patch:
diff --git a/testnet/stacks-node/Cargo.toml b/testnet/stacks-node/Cargo.toml
index 0b9b59a0e7..958820b491 100644
--- a/testnet/stacks-node/Cargo.toml
+++ b/testnet/stacks-node/Cargo.toml
@@ -50,6 +50,8 @@ tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 mutants = "0.0.3"
 tiny_http = "0.12.0"
 http-types = "2.12"
+tempfile = "3.3"
+mockito = "1.5"
 
 [[bin]]
 name = "stacks-node"
diff --git a/testnet/stacks-node/src/config.rs b/testnet/stacks-node/src/config.rs
index f4750c6242..d88330c87b 100644
--- a/testnet/stacks-node/src/config.rs
+++ b/testnet/stacks-node/src/config.rs
@@ -955,6 +955,18 @@ impl Config {
         })
     }
 
+    /// Returns the path working directory path, and ensures it exists.
+    pub fn get_working_dir(&self) -> PathBuf {
+        let path = PathBuf::from(&self.node.working_dir);
+        fs::create_dir_all(&path).unwrap_or_else(|_| {
+            panic!(
+                "Failed to create working directory at {}",
+                path.to_string_lossy()
+            )
+        });
+        path
+    }
+
     fn get_burnchain_path(&self) -> PathBuf {
         let mut path = PathBuf::from(&self.node.working_dir);
         path.push(&self.burnchain.mode);
diff --git a/testnet/stacks-node/src/event_dispatcher.rs b/testnet/stacks-node/src/event_dispatcher.rs
index faf437c444..43714f3573 100644
--- a/testnet/stacks-node/src/event_dispatcher.rs
+++ b/testnet/stacks-node/src/event_dispatcher.rs
@@ -16,6 +16,7 @@
 
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, HashSet};
+use std::path::PathBuf;
 use std::sync::mpsc::{channel, Receiver, Sender};
 use std::sync::Mutex;
 use std::thread::sleep;
@@ -25,6 +26,7 @@ use clarity::vm::analysis::contract_interface_builder::build_contract_interface;
 use clarity::vm::costs::ExecutionCost;
 use clarity::vm::events::{FTEventType, NFTEventType, STXEventType};
 use clarity::vm::types::{AssetIdentifier, QualifiedContractIdentifier, Value};
+use rusqlite::{params, Connection};
 use serde_json::json;
 use stacks::burnchains::{PoxConstants, Txid};
 use stacks::chainstate::burn::operations::BlockstackOperationType;
@@ -56,6 +58,7 @@ use stacks::net::http::HttpRequestContents;
 use stacks::net::httpcore::{send_http_request, StacksHttpRequest};
 use stacks::net::stackerdb::StackerDBEventDispatcher;
 use stacks::util::hash::to_hex;
+use stacks::util_lib::db::Error as db_error;
 use stacks_common::bitvec::BitVec;
 use stacks_common::codec::StacksMessageCodec;
 use stacks_common::types::chainstate::{BlockHeaderHash, BurnchainHeaderHash, StacksBlockId};
@@ -68,7 +71,12 @@ use super::config::{EventKeyType, EventObserverConfig};
 
 #[derive(Debug, Clone)]
 struct EventObserver {
+    /// Path to the database where pending payloads are stored. If `None`, then
+    /// the database is not used and events are not recoverable across restarts.
+    db_path: Option<PathBuf>,
+    /// URL to which events will be sent
     endpoint: String,
+    /// Timeout for sending events to this observer
     timeout: Duration,
 }
 
@@ -313,22 +321,106 @@ impl RewardSetEventPayload {
     }
 }
 
+#[cfg(test)]
+static TEST_EVENT_OBSERVER_SKIP_RETRY: std::sync::Mutex<Option<bool>> = std::sync::Mutex::new(None);
+
 impl EventObserver {
-    pub fn send_payload(&self, payload: &serde_json::Value, path: &str) {
+    fn init_db(db_path: &str) -> Result<Connection, db_error> {
+        let conn = Connection::open(db_path)?;
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS pending_payloads (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                url TEXT NOT NULL,
+                payload TEXT NOT NULL,
+                timeout INTEGER NOT NULL
+            )",
+            [],
+        )?;
+        Ok(conn)
+    }
+
+    fn insert_payload(
+        conn: &Connection,
+        url: &str,
+        payload: &serde_json::Value,
+        timeout: Duration,
+    ) -> Result<(), db_error> {
+        let payload_text = payload.to_string();
+        let timeout_ms: u64 = timeout.as_millis().try_into().expect("Timeout too large");
+        conn.execute(
+            "INSERT INTO pending_payloads (url, payload, timeout) VALUES (?1, ?2, ?3)",
+            params![url, payload_text, timeout_ms],
+        )?;
+        Ok(())
+    }
+
+    fn get_pending_payloads(
+        conn: &Connection,
+    ) -> Result<Vec<(i64, String, serde_json::Value, u64)>, db_error> {
+        let mut stmt =
+            conn.prepare("SELECT id, url, payload, timeout FROM pending_payloads ORDER BY id")?;
+        let payload_iter = stmt.query_and_then(
+            [],
+            |row| -> Result<(i64, String, serde_json::Value, u64), db_error> {
+                let id: i64 = row.get(0)?;
+                let url: String = row.get(1)?;
+                let payload_text: String = row.get(2)?;
+                let payload: serde_json::Value = serde_json::from_str(&payload_text)
+                    .map_err(|e| db_error::SerializationError(e))?;
+                let timeout_ms: u64 = row.get(3)?;
+                Ok((id, url, payload, timeout_ms))
+            },
+        )?;
+        payload_iter.collect()
+    }
+
+    fn delete_payload(conn: &Connection, id: i64) -> Result<(), db_error> {
+        conn.execute("DELETE FROM pending_payloads WHERE id = ?1", params![id])?;
+        Ok(())
+    }
+
+    fn process_pending_payloads(conn: &Connection) {
+        let pending_payloads = match Self::get_pending_payloads(conn) {
+            Ok(payloads) => payloads,
+            Err(e) => {
+                error!(
+                    "Event observer: failed to retrieve pending payloads from database";
+                    "error" => ?e
+                );
+                return;
+            }
+        };
+
+        for (id, url, payload, timeout_ms) in pending_payloads {
+            let timeout = Duration::from_millis(timeout_ms);
+            Self::send_payload_directly(&payload, &url, timeout);
+
+            #[cfg(test)]
+            if TEST_EVENT_OBSERVER_SKIP_RETRY
+                .lock()
+                .unwrap()
+                .unwrap_or(false)
+            {
+                warn!("Fault injection: delete_payload");
+                return;
+            }
+
+            if let Err(e) = Self::delete_payload(conn, id) {
+                error!(
+                    "Event observer: failed to delete pending payload from database";
+                    "error" => ?e
+                );
+            }
+        }
+    }
+
+    fn send_payload_directly(payload: &serde_json::Value, full_url: &str, timeout: Duration) {
         debug!(
-            "Event dispatcher: Sending payload"; "url" => %path, "payload" => ?payload
+            "Event dispatcher: Sending payload"; "url" => %full_url, "payload" => ?payload
         );
 
-        let url = {
-            let joined_components = if path.starts_with('/') {
-                format!("{}{}", &self.endpoint, path)
-            } else {
-                format!("{}/{}", &self.endpoint, path)
-            };
-            let url = format!("http://{}", joined_components);
-            Url::parse(&url)
-                .unwrap_or_else(|_| panic!("Event dispatcher: unable to parse {} as a URL", url))
-        };
+        let url = Url::parse(full_url)
+            .unwrap_or_else(|_| panic!("Event dispatcher: unable to parse {} as a URL", full_url));
 
         let host = url.host_str().expect("Invalid URL: missing host");
         let port = url.port_or_known_default().unwrap_or(80);
@@ -347,7 +439,7 @@ impl EventObserver {
             .unwrap_or_else(|_| panic!("FATAL: failed to encode infallible data as HTTP request"));
             request.add_header("Connection".into(), "close".into());
 
-            match send_http_request(host, port, request, self.timeout) {
+            match send_http_request(host, port, request, timeout) {
                 Ok(response) => {
                     if response.preamble().status_code == 200 {
                         debug!(
@@ -367,11 +459,71 @@ impl EventObserver {
                     );
                 }
             }
+
+            #[cfg(test)]
+            if TEST_EVENT_OBSERVER_SKIP_RETRY
+                .lock()
+                .unwrap()
+                .unwrap_or(false)
+            {
+                warn!("Fault injection: skipping retry of payload");
+                return;
+            }
+
             sleep(backoff);
             backoff *= 2;
         }
     }
 
+    fn new(working_dir: Option<PathBuf>, endpoint: String, timeout: Duration) -> Self {
+        let db_path = if let Some(mut db_path) = working_dir {
+            db_path.push("event_observers.sqlite");
+
+            Self::init_db(
+                db_path
+                    .to_str()
+                    .expect("Failed to convert chainstate path to string"),
+            )
+            .expect("Failed to initialize database for event observer");
+            Some(db_path)
+        } else {
+            None
+        };
+
+        EventObserver {
+            db_path,
+            endpoint,
+            timeout,
+        }
+    }
+
+    /// Send the payload to the given URL.
+    /// Before sending this payload, any pending payloads in the database will be sent first.
+    pub fn send_payload(&self, payload: &serde_json::Value, path: &str) {
+        // Construct the full URL
+        let url_str = if path.starts_with('/') {
+            format!("{}{}", &self.endpoint, path)
+        } else {
+            format!("{}/{}", &self.endpoint, path)
+        };
+        let full_url = format!("http://{}", url_str);
+
+        if let Some(db_path) = &self.db_path {
+            let conn =
+                Connection::open(db_path).expect("Failed to open database for event observer");
+
+            // Insert the new payload into the database
+            Self::insert_payload(&conn, &full_url, payload, self.timeout)
+                .expect("Failed to insert payload into event observer database");
+
+            // Process all pending payloads
+            Self::process_pending_payloads(&conn);
+        } else {
+            // No database, just send the payload
+            Self::send_payload_directly(payload, &full_url, self.timeout);
+        }
+    }
+
     fn make_new_mempool_txs_payload(transactions: Vec<StacksTransaction>) -> serde_json::Value {
         let raw_txs = transactions
             .into_iter()
@@ -1403,12 +1555,13 @@ impl EventDispatcher {
         }
     }
 
-    pub fn register_observer(&mut self, conf: &EventObserverConfig) {
+    pub fn register_observer(&mut self, conf: &EventObserverConfig, working_dir: PathBuf) {
         info!("Registering event observer at: {}", conf.endpoint);
-        let event_observer = EventObserver {
-            endpoint: conf.endpoint.clone(),
-            timeout: Duration::from_millis(conf.timeout_ms),
-        };
+        let event_observer = EventObserver::new(
+            Some(working_dir),
+            conf.endpoint.clone(),
+            Duration::from_millis(conf.timeout_ms),
+        );
 
         let observer_index = self.registered_observers.len() as u16;
 
@@ -1492,16 +1645,14 @@ mod test {
     use stacks::util::secp256k1::MessageSignature;
     use stacks_common::bitvec::BitVec;
     use stacks_common::types::chainstate::{BurnchainHeaderHash, StacksBlockId};
+    use tempfile::tempdir;
     use tiny_http::{Method, Response, Server, StatusCode};
 
     use super::*;
 
     #[test]
     fn build_block_processed_event() {
-        let observer = EventObserver {
-            endpoint: "nowhere".to_string(),
-            timeout: Duration::from_secs(3),
-        };
+        let observer = EventObserver::new(None, "nowhere".to_string(), Duration::from_secs(3));
 
         let filtered_events = vec![];
         let block = StacksBlock::genesis_block();
@@ -1559,10 +1710,7 @@ mod test {
 
     #[test]
     fn test_block_processed_event_nakamoto() {
-        let observer = EventObserver {
-            endpoint: "nowhere".to_string(),
-            timeout: Duration::from_secs(3),
-        };
+        let observer = EventObserver::new(None, "nowhere".to_string(), Duration::from_secs(3));
 
         let filtered_events = vec![];
         let mut block_header = NakamotoBlockHeader::empty();
@@ -1679,6 +1827,231 @@ mod test {
         listener.local_addr().unwrap().port()
     }
 
+    #[test]
+    fn test_init_db() {
+        let dir = tempdir().unwrap();
+        let db_path = dir.path().join("test_init_db.sqlite");
+        let db_path_str = db_path.to_str().unwrap();
+
+        // Call init_db
+        let conn_result = EventObserver::init_db(db_path_str);
+        assert!(conn_result.is_ok(), "Failed to initialize the database");
+
+        // Check that the database file exists
+        assert!(db_path.exists(), "Database file was not created");
+
+        // Check that the table exists
+        let conn = conn_result.unwrap();
+        let mut stmt = conn
+            .prepare(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name='pending_payloads'",
+            )
+            .unwrap();
+        let table_exists = stmt.exists([]).unwrap();
+        assert!(table_exists, "Table 'pending_payloads' does not exist");
+    }
+
+    #[test]
+    fn test_insert_and_get_pending_payloads() {
+        let dir = tempdir().unwrap();
+        let db_path = dir.path().join("test_payloads.sqlite");
+        let db_path_str = db_path.to_str().unwrap();
+
+        let conn = EventObserver::init_db(db_path_str).expect("Failed to initialize the database");
+
+        let url = "http://example.com/api";
+        let payload = json!({"key": "value"});
+        let timeout = Duration::from_secs(5);
+
+        // Insert payload
+        let insert_result = EventObserver::insert_payload(&conn, url, &payload, timeout);
+        assert!(insert_result.is_ok(), "Failed to insert payload");
+
+        // Get pending payloads
+        let pending_payloads =
+            EventObserver::get_pending_payloads(&conn).expect("Failed to get pending payloads");
+        assert_eq!(pending_payloads.len(), 1, "Expected one pending payload");
+
+        let (_id, retrieved_url, retrieved_payload, timeout_ms) = &pending_payloads[0];
+        assert_eq!(retrieved_url, url, "URL does not match");
+        assert_eq!(retrieved_payload, &payload, "Payload does not match");
+        assert_eq!(
+            *timeout_ms,
+            timeout.as_millis() as u64,
+            "Timeout does not match"
+        );
+    }
+
+    #[test]
+    fn test_delete_payload() {
+        let dir = tempdir().unwrap();
+        let db_path = dir.path().join("test_delete_payload.sqlite");
+        let db_path_str = db_path.to_str().unwrap();
+
+        let conn = EventObserver::init_db(db_path_str).expect("Failed to initialize the database");
+
+        let url = "http://example.com/api";
+        let payload = json!({"key": "value"});
+        let timeout = Duration::from_secs(5);
+
+        // Insert payload
+        EventObserver::insert_payload(&conn, url, &payload, timeout)
+            .expect("Failed to insert payload");
+
+        // Get pending payloads
+        let pending_payloads =
+            EventObserver::get_pending_payloads(&conn).expect("Failed to get pending payloads");
+        assert_eq!(pending_payloads.len(), 1, "Expected one pending payload");
+
+        let (id, _, _, _) = pending_payloads[0];
+
+        // Delete payload
+        let delete_result = EventObserver::delete_payload(&conn, id);
+        assert!(delete_result.is_ok(), "Failed to delete payload");
+
+        // Verify that the pending payloads list is empty
+        let pending_payloads =
+            EventObserver::get_pending_payloads(&conn).expect("Failed to get pending payloads");
+        assert_eq!(pending_payloads.len(), 0, "Expected no pending payloads");
+    }
+
+    #[test]
+    fn test_process_pending_payloads() {
+        use mockito::Matcher;
+
+        let dir = tempdir().unwrap();
+        let db_path = dir.path().join("test_process_payloads.sqlite");
+        let db_path_str = db_path.to_str().unwrap();
+
+        let conn = EventObserver::init_db(db_path_str).expect("Failed to initialize the database");
+
+        let payload = json!({"key": "value"});
+        let timeout = Duration::from_secs(5);
+
+        // Create a mock server
+        let mut server = mockito::Server::new();
+        let _m = server
+            .mock("POST", "/api")
+            .match_header("content-type", Matcher::Regex("application/json.*".into()))
+            .match_body(Matcher::Json(payload.clone()))
+            .with_status(200)
+            .create();
+
+        let url = &format!("{}/api", &server.url());
+
+        // Insert payload
+        EventObserver::insert_payload(&conn, url, &payload, timeout)
+            .expect("Failed to insert payload");
+
+        // Process pending payloads
+        EventObserver::process_pending_payloads(&conn);
+
+        // Verify that the pending payloads list is empty
+        let pending_payloads =
+            EventObserver::get_pending_payloads(&conn).expect("Failed to get pending payloads");
+        assert_eq!(pending_payloads.len(), 0, "Expected no pending payloads");
+
+        // Verify that the mock was called
+        _m.assert();
+    }
+
+    #[test]
+    fn test_new_event_observer_with_db() {
+        let dir = tempdir().unwrap();
+        let working_dir = dir.path().to_path_buf();
+
+        let endpoint = "http://example.com".to_string();
+        let timeout = Duration::from_secs(5);
+
+        let observer = EventObserver::new(Some(working_dir.clone()), endpoint.clone(), timeout);
+
+        // Verify fields
+        assert_eq!(observer.endpoint, endpoint);
+        assert_eq!(observer.timeout, timeout);
+
+        // Verify that the database was initialized
+        let mut db_path = working_dir;
+        db_path.push("event_observers.sqlite");
+        assert!(db_path.exists(), "Database file was not created");
+    }
+
+    #[test]
+    fn test_new_event_observer_without_db() {
+        let endpoint = "http://example.com".to_string();
+        let timeout = Duration::from_secs(5);
+
+        let observer = EventObserver::new(None, endpoint.clone(), timeout);
+
+        // Verify fields
+        assert_eq!(observer.endpoint, endpoint);
+        assert_eq!(observer.timeout, timeout);
+        assert!(observer.db_path.is_none(), "Expected db_path to be None");
+    }
+
+    #[test]
+    fn test_send_payload_with_db() {
+        use mockito::Matcher;
+
+        let dir = tempdir().unwrap();
+        let working_dir = dir.path().to_path_buf();
+        let payload = json!({"key": "value"});
+
+        // Create a mock server
+        let mut server = mockito::Server::new();
+        let _m = server
+            .mock("POST", "/test")
+            .match_header("content-type", Matcher::Regex("application/json.*".into()))
+            .match_body(Matcher::Json(payload.clone()))
+            .with_status(200)
+            .create();
+
+        let endpoint = server.url().strip_prefix("http://").unwrap().to_string();
+        let timeout = Duration::from_secs(5);
+
+        let observer = EventObserver::new(Some(working_dir.clone()), endpoint, timeout);
+
+        // Call send_payload
+        observer.send_payload(&payload, "/test");
+
+        // Verify that the payload was sent and database is empty
+        _m.assert();
+
+        // Verify that the database is empty
+        let db_path = observer.db_path.unwrap();
+        let db_path_str = db_path.to_str().unwrap();
+        let conn = Connection::open(db_path_str).expect("Failed to open database");
+        let pending_payloads =
+            EventObserver::get_pending_payloads(&conn).expect("Failed to get pending payloads");
+        assert_eq!(pending_payloads.len(), 0, "Expected no pending payloads");
+    }
+
+    #[test]
+    fn test_send_payload_without_db() {
+        use mockito::Matcher;
+
+        let timeout = Duration::from_secs(5);
+        let payload = json!({"key": "value"});
+
+        // Create a mock server
+        let mut server = mockito::Server::new();
+        let _m = server
+            .mock("POST", "/test")
+            .match_header("content-type", Matcher::Regex("application/json.*".into()))
+            .match_body(Matcher::Json(payload.clone()))
+            .with_status(200)
+            .create();
+
+        let endpoint = server.url().strip_prefix("http://").unwrap().to_string();
+
+        let observer = EventObserver::new(None, endpoint, timeout);
+
+        // Call send_payload
+        observer.send_payload(&payload, "/test");
+
+        // Verify that the payload was sent
+        _m.assert();
+    }
+
     #[test]
     fn test_send_payload_success() {
         let port = get_random_port();
@@ -1701,10 +2074,8 @@ mod test {
             tx.send(()).unwrap();
         });
 
-        let observer = EventObserver {
-            endpoint: format!("127.0.0.1:{}", port),
-            timeout: Duration::from_secs(3),
-        };
+        let observer =
+            EventObserver::new(None, format!("127.0.0.1:{}", port), Duration::from_secs(3));
 
         let payload = json!({"key": "value"});
 
@@ -1752,10 +2123,8 @@ mod test {
             }
         });
 
-        let observer = EventObserver {
-            endpoint: format!("127.0.0.1:{}", port),
-            timeout: Duration::from_secs(3),
-        };
+        let observer =
+            EventObserver::new(None, format!("127.0.0.1:{}", port), Duration::from_secs(3));
 
         let payload = json!({"key": "value"});
 
@@ -1799,10 +2168,7 @@ mod test {
             }
         });
 
-        let observer = EventObserver {
-            endpoint: format!("127.0.0.1:{}", port),
-            timeout,
-        };
+        let observer = EventObserver::new(None, format!("127.0.0.1:{}", port), timeout);
 
         let payload = json!({"key": "value"});
 
@@ -1830,4 +2196,99 @@ mod test {
         rx.recv_timeout(Duration::from_secs(5))
             .expect("Server did not receive request in time");
     }
+
+    #[test]
+    fn test_send_payload_with_db_force_restart() {
+        let port = get_random_port();
+        let timeout = Duration::from_secs(3);
+        let dir = tempdir().unwrap();
+        let working_dir = dir.path().to_path_buf();
+
+        // Set up a channel to notify when the server has processed the request
+        let (tx, rx) = channel();
+
+        info!("Starting mock server on port {}", port);
+        // Start a mock server in a separate thread
+        let server = Server::http(format!("127.0.0.1:{}", port)).unwrap();
+        thread::spawn(move || {
+            let mut attempt = 0;
+            let mut _request_holder = None;
+            while let Ok(mut request) = server.recv() {
+                attempt += 1;
+                match attempt {
+                    1 => {
+                        debug!("Mock server received request attempt 1");
+                        // Do not reply, forcing the sender to timeout and retry,
+                        // but don't drop the request or it will receive a 500 error,
+                        _request_holder = Some(request);
+                    }
+                    2 => {
+                        debug!("Mock server received request attempt 2");
+
+                        // Verify the payload
+                        let mut payload = String::new();
+                        request.as_reader().read_to_string(&mut payload).unwrap();
+                        let expected_payload = r#"{"key":"value"}"#;
+                        assert_eq!(payload, expected_payload);
+
+                        // Simulate a successful response on the second attempt
+                        let response = Response::from_string("HTTP/1.1 200 OK");
+                        request.respond(response).unwrap();
+                    }
+                    3 => {
+                        debug!("Mock server received request attempt 3");
+
+                        // Verify the payload
+                        let mut payload = String::new();
+                        request.as_reader().read_to_string(&mut payload).unwrap();
+                        let expected_payload = r#"{"key":"value2"}"#;
+                        assert_eq!(payload, expected_payload);
+
+                        // Simulate a successful response on the second attempt
+                        let response = Response::from_string("HTTP/1.1 200 OK");
+                        request.respond(response).unwrap();
+
+                        // When we receive attempt 3 (message 1, re-sent message 1, message 2),
+                        // notify the test that the request was processed successfully
+                        tx.send(()).unwrap();
+                        break;
+                    }
+                    _ => panic!("Unexpected request attempt"),
+                }
+            }
+        });
+
+        let observer = EventObserver::new(
+            Some(working_dir.clone()),
+            format!("127.0.0.1:{}", port),
+            timeout,
+        );
+
+        let payload = json!({"key": "value"});
+        let payload2 = json!({"key": "value2"});
+
+        // Disable retrying so that it sends the payload only once
+        // and that payload will be ignored by the test server.
+        TEST_EVENT_OBSERVER_SKIP_RETRY.lock().unwrap().replace(true);
+
+        info!("Sending payload 1");
+
+        // Send the payload
+        observer.send_payload(&payload, "/test");
+
+        // Re-enable retrying
+        TEST_EVENT_OBSERVER_SKIP_RETRY
+            .lock()
+            .unwrap()
+            .replace(false);
+
+        info!("Sending payload 2");
+
+        // Send another payload
+        observer.send_payload(&payload2, "/test");
+
+        // Wait for the server to process the requests
+        rx.recv_timeout(Duration::from_secs(5))
+            .expect("Server did not receive request in time");
+    }
 }
diff --git a/testnet/stacks-node/src/node.rs b/testnet/stacks-node/src/node.rs
index 39095a51d5..1895912ba5 100644
--- a/testnet/stacks-node/src/node.rs
+++ b/testnet/stacks-node/src/node.rs
@@ -340,7 +340,7 @@ impl Node {
         let mut event_dispatcher = EventDispatcher::new();
 
         for observer in &config.events_observers {
-            event_dispatcher.register_observer(observer);
+            event_dispatcher.register_observer(observer, config.get_working_dir());
         }
 
         let burnchain_config = config.get_burnchain();
diff --git a/testnet/stacks-node/src/run_loop/nakamoto.rs b/testnet/stacks-node/src/run_loop/nakamoto.rs
index 6e3222d99d..04afdd79ee 100644
--- a/testnet/stacks-node/src/run_loop/nakamoto.rs
+++ b/testnet/stacks-node/src/run_loop/nakamoto.rs
@@ -93,7 +93,7 @@ impl RunLoop {
 
         let mut event_dispatcher = EventDispatcher::new();
         for observer in config.events_observers.iter() {
-            event_dispatcher.register_observer(observer);
+            event_dispatcher.register_observer(observer, config.get_working_dir());
         }
 
         Self {
diff --git a/testnet/stacks-node/src/run_loop/neon.rs b/testnet/stacks-node/src/run_loop/neon.rs
index 331e7e597c..a18a61988b 100644
--- a/testnet/stacks-node/src/run_loop/neon.rs
+++ b/testnet/stacks-node/src/run_loop/neon.rs
@@ -236,7 +236,7 @@ impl RunLoop {
 
         let mut event_dispatcher = EventDispatcher::new();
         for observer in config.events_observers.iter() {
-            event_dispatcher.register_observer(observer);
+            event_dispatcher.register_observer(observer, config.get_working_dir());
         }
 
         Self {