Compare commits

..
Author SHA1 Message Date
William Casarin c12da6a229 Command result notices
This introduces a new notice-like event with more structure so that clients
can know if an event is sucessfully written to the database. Clients
can't really do much with the current NOTICE messages, but with these
structured result messages, it can know if an event was sucessfully
saved.

Whenever there is an error and we have an id available (event id, sub
id, etc) we return these structured OK events instead.

Example:

When saving the following event:
    ["EVENT",{"id": "event_id" }]

The server will now return events in the following format:
    ["OK", event_id, "true|false", message]

For example, on a successful save:
    ["OK", "event_id", "true"]

If we already have the event:
    ["OK", "event_id", "true", "duplicate"]

If the event is rejected:
    ["OK", "event_id", "false", "you are blocked"]

If a subscription fails:
    ["OK", "sub_id", "false", "Too many subscriptions"]

NIP coming soon!
2022-11-09 07:44:39 -06:00
16 changed files with 249 additions and 599 deletions
Generated
+105 -112
View File
@@ -27,9 +27,9 @@ dependencies = [
[[package]] [[package]]
name = "aho-corasick" name = "aho-corasick"
version = "0.7.20" version = "0.7.19"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc936419f96fa211c1b9166887b38e5e40b19958e5b895be7c1f93adec7071ac" checksum = "b4f55bd91a0978cbfd91c457a164bab8b4001c833b7f323132c0a4e1922dd44e"
dependencies = [ dependencies = [
"memchr", "memchr",
] ]
@@ -54,9 +54,9 @@ dependencies = [
[[package]] [[package]]
name = "anyhow" name = "anyhow"
version = "1.0.67" version = "1.0.66"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7724808837b77f4b4de9d283820f9d98bcf496d5692934b857a2399d31ff22e6" checksum = "216261ddc8289130e551ddcd5ce8a064710c0d064a4d2895c67151c92b5443f6"
[[package]] [[package]]
name = "async-stream" name = "async-stream"
@@ -81,9 +81,9 @@ dependencies = [
[[package]] [[package]]
name = "async-trait" name = "async-trait"
version = "0.1.60" version = "0.1.58"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "677d1d8ab452a3936018a687b20e6f7cf5363d713b732b8884001317b0e48aa3" checksum = "1e805d94e6b5001b651426cf4cd446b1ab5f319d27bab5c644f61de0a804360c"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@@ -107,9 +107,9 @@ checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
[[package]] [[package]]
name = "axum" name = "axum"
version = "0.6.1" version = "0.5.17"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "08b108ad2665fa3f6e6a517c3d80ec3e77d224c47d605167aefaa5d7ef97fa48" checksum = "acee9fd5073ab6b045a275b3e709c163dd36c90685219cb21804a147b58dba43"
dependencies = [ dependencies = [
"async-trait", "async-trait",
"axum-core", "axum-core",
@@ -125,9 +125,9 @@ dependencies = [
"mime", "mime",
"percent-encoding", "percent-encoding",
"pin-project-lite", "pin-project-lite",
"rustversion",
"serde", "serde",
"sync_wrapper", "sync_wrapper",
"tokio",
"tower", "tower",
"tower-http", "tower-http",
"tower-layer", "tower-layer",
@@ -136,9 +136,9 @@ dependencies = [
[[package]] [[package]]
name = "axum-core" name = "axum-core"
version = "0.3.0" version = "0.2.9"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "79b8558f5a0581152dc94dcd289132a1d377494bdeafcd41869b3258e3e2ad92" checksum = "37e5939e02c56fecd5c017c37df4238c0a839fa76b7f97acdd7efb804fd181cc"
dependencies = [ dependencies = [
"async-trait", "async-trait",
"bytes", "bytes",
@@ -146,7 +146,6 @@ dependencies = [
"http", "http",
"http-body", "http-body",
"mime", "mime",
"rustversion",
"tower-layer", "tower-layer",
"tower-service", "tower-service",
] ]
@@ -195,15 +194,15 @@ checksum = "14c189c53d098945499cdfa7ecc63567cf3886b3332b312a5b4585d8d3a6a610"
[[package]] [[package]]
name = "bytes" name = "bytes"
version = "1.3.0" version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dfb24e866b15a1af2a1b663f10c6b6b8f397a84aadb828f12e5b289ec23a3a3c" checksum = "ec8a7b6a70fde80372154c65702f00a0f56f3e1c36abbc6c440484be248856db"
[[package]] [[package]]
name = "cc" name = "cc"
version = "1.0.78" version = "1.0.74"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a20104e2335ce8a659d6dd92a51a767a0c062599c73b343fd152cb401e828c3d" checksum = "581f5dba903aac52ea3feb5ec4810848460ee833876f1f9b0fdeab1f19091574"
[[package]] [[package]]
name = "cfg-if" name = "cfg-if"
@@ -213,9 +212,9 @@ checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
[[package]] [[package]]
name = "chrono" name = "chrono"
version = "0.4.23" version = "0.4.22"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "16b0a3d9ed01224b22057780a37bb8c5dbfe1be8ba48678e7bf57ec4b385411f" checksum = "bfd4d1b31faaa3a89d7934dbded3111da0d2ef28e3ebccdb4f0179f5929d1ef1"
dependencies = [ dependencies = [
"iana-time-zone", "iana-time-zone",
"num-integer", "num-integer",
@@ -363,9 +362,9 @@ dependencies = [
[[package]] [[package]]
name = "crossbeam-utils" name = "crossbeam-utils"
version = "0.8.14" version = "0.8.12"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4fb766fa798726286dbbb842f174001dab8abc7b627a1dd86e0b7222a95d929f" checksum = "edbafec5fa1f196ca66527c1b12c2ec4745ca14b50f1ad8f9f6f720b55d11fac"
dependencies = [ dependencies = [
"cfg-if", "cfg-if",
] ]
@@ -382,9 +381,9 @@ dependencies = [
[[package]] [[package]]
name = "cxx" name = "cxx"
version = "1.0.84" version = "1.0.80"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "27874566aca772cb515af4c6e997b5fe2119820bca447689145e39bb734d19a0" checksum = "6b7d4e43b25d3c994662706a1d4fcfc32aaa6afd287502c111b237093bb23f3a"
dependencies = [ dependencies = [
"cc", "cc",
"cxxbridge-flags", "cxxbridge-flags",
@@ -394,9 +393,9 @@ dependencies = [
[[package]] [[package]]
name = "cxx-build" name = "cxx-build"
version = "1.0.84" version = "1.0.80"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e7bb951f2523a49533003656a72121306b225ec16a49a09dc6b0ba0d6f3ec3c0" checksum = "84f8829ddc213e2c1368e51a2564c552b65a8cb6a28f31e576270ac81d5e5827"
dependencies = [ dependencies = [
"cc", "cc",
"codespan-reporting", "codespan-reporting",
@@ -409,15 +408,15 @@ dependencies = [
[[package]] [[package]]
name = "cxxbridge-flags" name = "cxxbridge-flags"
version = "1.0.84" version = "1.0.80"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "be778b6327031c1c7b61dd2e48124eee5361e6aa76b8de93692f011b08870ab4" checksum = "e72537424b474af1460806647c41d4b6d35d09ef7fe031c5c2fa5766047cc56a"
[[package]] [[package]]
name = "cxxbridge-macro" name = "cxxbridge-macro"
version = "1.0.84" version = "1.0.80"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7b8a2b87662fe5a0a0b38507756ab66aff32638876a0866e5a5fc82ceb07ee49" checksum = "309e4fb93eed90e1e14bea0da16b209f81813ba9fc7830c20ed151dd7bc0a4d7"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@@ -439,9 +438,9 @@ dependencies = [
[[package]] [[package]]
name = "digest" name = "digest"
version = "0.10.6" version = "0.10.5"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8168378f4e5023e7218c89c891c0fd8ecdb5e5e4f18cb78f38cf245dd021e76f" checksum = "adfbc57365a37acbd2ebf2b64d7e69bb766e2fea813521ed536f5d0520dcf86c"
dependencies = [ dependencies = [
"block-buffer", "block-buffer",
"crypto-common", "crypto-common",
@@ -485,9 +484,9 @@ dependencies = [
[[package]] [[package]]
name = "flate2" name = "flate2"
version = "1.0.25" version = "1.0.24"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a8a2db397cb1c8772f31494cb8917e48cd1e64f0fa7efac59fbd741a0a8ce841" checksum = "f82b0f4c27ad9f8bfd1f3208d882da2b09c301bc1c828fd3a00d0216d2fbbff6"
dependencies = [ dependencies = [
"crc32fast", "crc32fast",
"miniz_oxide", "miniz_oxide",
@@ -790,9 +789,9 @@ checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"
[[package]] [[package]]
name = "hyper" name = "hyper"
version = "0.14.23" version = "0.14.22"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "034711faac9d2166cb1baf1a2fb0b60b1f277f8492fd72176c17f3515e1abd3c" checksum = "abfba89e19b959ca163c7752ba59d737c1ceea53a5d31a149c805446fc958064"
dependencies = [ dependencies = [
"bytes", "bytes",
"futures-channel", "futures-channel",
@@ -873,9 +872,9 @@ dependencies = [
[[package]] [[package]]
name = "indexmap" name = "indexmap"
version = "1.9.2" version = "1.9.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1885e79c1fc4b10f0e172c475f458b7f7b93061064d98c3293e98c5ba0c8b399" checksum = "10a35a97730320ffe8e2d410b5d3b69279b98d2c14bdb8b70ea89ecf7888d41e"
dependencies = [ dependencies = [
"autocfg 1.1.0", "autocfg 1.1.0",
"hashbrown 0.12.3", "hashbrown 0.12.3",
@@ -901,9 +900,9 @@ dependencies = [
[[package]] [[package]]
name = "itoa" name = "itoa"
version = "1.0.5" version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fad582f4b9e86b6caa621cabeb0963332d92eea04729ab12892c2533951e6440" checksum = "4217ad341ebadf8d8e724e264f13e593e0648f5b3e94b3896a5df283be015ecc"
[[package]] [[package]]
name = "js-sys" name = "js-sys"
@@ -933,9 +932,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
[[package]] [[package]]
name = "libc" name = "libc"
version = "0.2.138" version = "0.2.137"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "db6d7e329c562c5dfab7a46a2afabc8b987ab9a4834c9d1ca04dc54c1546cef8" checksum = "fc7fcc620a3bff7cdd7a365be3376c97191aeaccc2a603e600951e452615bf89"
[[package]] [[package]]
name = "libsqlite3-sys" name = "libsqlite3-sys"
@@ -950,9 +949,9 @@ dependencies = [
[[package]] [[package]]
name = "link-cplusplus" name = "link-cplusplus"
version = "1.0.8" version = "1.0.7"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ecd207c9c713c34f95a097a5b029ac2ce6010530c7b49d7fea24d977dede04f5" checksum = "9272ab7b96c9046fbc5bc56c06c117cb639fe2d509df0c421cad82d2915cf369"
dependencies = [ dependencies = [
"cc", "cc",
] ]
@@ -1011,9 +1010,9 @@ dependencies = [
[[package]] [[package]]
name = "matchit" name = "matchit"
version = "0.7.0" version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b87248edafb776e59e6ee64a79086f65890d3510f2c656c000bf2a7e8a0aea40" checksum = "73cbba799671b762df5a175adf59ce145165747bb891505c43d09aefbbf38beb"
[[package]] [[package]]
name = "memchr" name = "memchr"
@@ -1035,9 +1034,9 @@ checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
[[package]] [[package]]
name = "miniz_oxide" name = "miniz_oxide"
version = "0.6.2" version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b275950c28b37e794e8c55d88aeb5e139d0ce23fdbbeda68f8d7174abdf9e8fa" checksum = "96590ba8f175222643a85693f33d26e9c8a015f599c216509b1a6894af675d34"
dependencies = [ dependencies = [
"adler", "adler",
] ]
@@ -1096,7 +1095,7 @@ checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21"
[[package]] [[package]]
name = "nostr-rs-relay" name = "nostr-rs-relay"
version = "0.7.12" version = "0.7.1"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"bitcoin_hashes", "bitcoin_hashes",
@@ -1225,9 +1224,9 @@ checksum = "86f0b0d4bf799edbc74508c1e8bf170ff5f41238e5f8225603ca7caaae2b7860"
[[package]] [[package]]
name = "openssl" name = "openssl"
version = "0.10.44" version = "0.10.42"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29d971fd5722fec23977260f6e81aa67d2f22cadbdc2aa049f1022d9a3be1566" checksum = "12fc0523e3bd51a692c8850d075d74dc062ccf251c0110668cbd921917118a13"
dependencies = [ dependencies = [
"bitflags", "bitflags",
"cfg-if", "cfg-if",
@@ -1257,9 +1256,9 @@ checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
[[package]] [[package]]
name = "openssl-sys" name = "openssl-sys"
version = "0.9.79" version = "0.9.77"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5454462c0eced1e97f2ec09036abc8da362e66802f66fd20f86854d9d8cbcbc4" checksum = "b03b84c3b2d099b81f0953422b4d4ad58761589d0229b5506356afca05a3670a"
dependencies = [ dependencies = [
"autocfg 1.1.0", "autocfg 1.1.0",
"cc", "cc",
@@ -1290,9 +1289,9 @@ dependencies = [
[[package]] [[package]]
name = "parking_lot_core" name = "parking_lot_core"
version = "0.9.5" version = "0.9.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7ff9f3fef3968a3ec5945535ed654cb38ff72d7495a25619e2247fb15a2ed9ba" checksum = "4dc9e0dc2adc1c69d09143aff38d3d30c5c3f0df0dad82e6d25547af174ebec0"
dependencies = [ dependencies = [
"cfg-if", "cfg-if",
"libc", "libc",
@@ -1326,9 +1325,9 @@ checksum = "478c572c3d73181ff3c2539045f6eb99e5491218eae919370993b890cdbdd98e"
[[package]] [[package]]
name = "pest" name = "pest"
version = "2.5.1" version = "2.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc8bed3549e0f9b0a2a78bf7c0018237a2cdf085eecbbc048e52612438e4e9d0" checksum = "a528564cc62c19a7acac4d81e01f39e53e25e17b934878f4c6d25cc2836e62f8"
dependencies = [ dependencies = [
"thiserror", "thiserror",
"ucd-trie", "ucd-trie",
@@ -1336,9 +1335,9 @@ dependencies = [
[[package]] [[package]]
name = "pest_derive" name = "pest_derive"
version = "2.5.1" version = "2.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cdc078600d06ff90d4ed238f0119d84ab5d43dbaad278b0e33a8820293b32344" checksum = "d5fd9bc6500181952d34bd0b2b0163a54d794227b498be0b7afa7698d0a7b18f"
dependencies = [ dependencies = [
"pest", "pest",
"pest_generator", "pest_generator",
@@ -1346,9 +1345,9 @@ dependencies = [
[[package]] [[package]]
name = "pest_generator" name = "pest_generator"
version = "2.5.1" version = "2.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "28a1af60b1c4148bb269006a750cff8e2ea36aff34d2d96cf7be0b14d1bed23c" checksum = "d2610d5ac5156217b4ff8e46ddcef7cdf44b273da2ac5bca2ecbfa86a330e7c4"
dependencies = [ dependencies = [
"pest", "pest",
"pest_meta", "pest_meta",
@@ -1359,9 +1358,9 @@ dependencies = [
[[package]] [[package]]
name = "pest_meta" name = "pest_meta"
version = "2.5.1" version = "2.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fec8605d59fc2ae0c6c1aefc0c7c7a9769732017c0ce07f7a9cfffa7b4404f20" checksum = "824749bf7e21dd66b36fbe26b3f45c713879cccd4a009a917ab8e045ca8246fe"
dependencies = [ dependencies = [
"once_cell", "once_cell",
"pest", "pest",
@@ -1414,18 +1413,18 @@ checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"
[[package]] [[package]]
name = "proc-macro2" name = "proc-macro2"
version = "1.0.48" version = "1.0.47"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e9d89e5dba24725ae5678020bf8f1357a9aa7ff10736b551adbcd3f8d17d766f" checksum = "5ea3d908b0e36316caf9e9e2c4625cdde190a7e6f440d794667ed17a1855e725"
dependencies = [ dependencies = [
"unicode-ident", "unicode-ident",
] ]
[[package]] [[package]]
name = "prost" name = "prost"
version = "0.11.3" version = "0.11.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c0b18e655c21ff5ac2084a5ad0611e827b3f92badf79f4910b5a5c58f4d87ff0" checksum = "a0841812012b2d4a6145fae9a6af1534873c32aa67fff26bd09f8fa42c83f95a"
dependencies = [ dependencies = [
"bytes", "bytes",
"prost-derive", "prost-derive",
@@ -1472,9 +1471,9 @@ dependencies = [
[[package]] [[package]]
name = "quote" name = "quote"
version = "1.0.22" version = "1.0.21"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "556d0f47a940e895261e77dc200d5eadfc6ef644c179c6f5edfc105e3a2292c8" checksum = "bbe448f377a7d6961e30f5955f9b8d106c3f5e449d493ee1b125c1d43c2b5179"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
] ]
@@ -1665,9 +1664,9 @@ dependencies = [
[[package]] [[package]]
name = "regex" name = "regex"
version = "1.7.0" version = "1.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e076559ef8e241f2ae3479e36f97bd5741c0330689e217ad51ce2c76808b868a" checksum = "4c4eb3267174b8c6c2f654116623910a0fef09c4753f8dd83db29c48a0df988b"
dependencies = [ dependencies = [
"aho-corasick", "aho-corasick",
"memchr", "memchr",
@@ -1685,9 +1684,9 @@ dependencies = [
[[package]] [[package]]
name = "regex-syntax" name = "regex-syntax"
version = "0.6.28" version = "0.6.27"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "456c603be3e8d448b072f410900c09faf164fbce2d480456f50eea6e25f9c848" checksum = "a3f87b73ce11b1619a3c6332f45341e0047173771e8b8b73f87bfeefb7b56244"
[[package]] [[package]]
name = "remove_dir_all" name = "remove_dir_all"
@@ -1735,16 +1734,10 @@ dependencies = [
] ]
[[package]] [[package]]
name = "rustversion" name = "ryu"
version = "1.0.11" version = "1.0.11"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5583e89e108996506031660fe09baa5011b9dd0341b89029313006d1fb508d70" checksum = "4501abdff3ae82a1c1b477a17252eb69cee9e66eb915c1abaa4f44d873df9f09"
[[package]]
name = "ryu"
version = "1.0.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7b4b9743ed687d4b4bcedf9ff5eaa7398495ae14e61cba0a295704edbc7decde"
[[package]] [[package]]
name = "schannel" name = "schannel"
@@ -1773,9 +1766,9 @@ checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"
[[package]] [[package]]
name = "scratch" name = "scratch"
version = "1.0.3" version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ddccb15bcce173023b3fedd9436f882a0739b8dfb45e4f6b6002bee5929f61b2" checksum = "9c8132065adcfd6e02db789d9285a0deb2f3fcb04002865ab67d5fb103533898"
[[package]] [[package]]
name = "secp256k1" name = "secp256k1"
@@ -1823,18 +1816,18 @@ dependencies = [
[[package]] [[package]]
name = "serde" name = "serde"
version = "1.0.151" version = "1.0.147"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "97fed41fc1a24994d044e6db6935e69511a1153b52c15eb42493b26fa87feba0" checksum = "d193d69bae983fc11a79df82342761dfbf28a99fc8d203dca4c3c1b590948965"
dependencies = [ dependencies = [
"serde_derive", "serde_derive",
] ]
[[package]] [[package]]
name = "serde_derive" name = "serde_derive"
version = "1.0.151" version = "1.0.147"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "255abe9a125a985c05190d687b320c12f9b1f0b99445e608c21ba0782c719ad8" checksum = "4f1d362ca8fc9c3e3a7484440752472d68a6caa98f1ab81d99b5dfe517cec852"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@@ -1843,9 +1836,9 @@ dependencies = [
[[package]] [[package]]
name = "serde_json" name = "serde_json"
version = "1.0.90" version = "1.0.87"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8778cc0b528968fe72abec38b5db5a20a70d148116cd9325d2bc5f5180ca3faf" checksum = "6ce777b7b150d76b9cf60d28b55f5847135a003f7d7350c6be7a773508ce7d45"
dependencies = [ dependencies = [
"indexmap", "indexmap",
"itoa", "itoa",
@@ -1855,9 +1848,9 @@ dependencies = [
[[package]] [[package]]
name = "sha-1" name = "sha-1"
version = "0.10.1" version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f5058ada175748e33390e40e872bd0fe59a19f265d0158daa551c5a88a76009c" checksum = "028f48d513f9678cda28f6e4064755b3fbb2af6acd672f2c209b62323f7aea0f"
dependencies = [ dependencies = [
"cfg-if", "cfg-if",
"cpufeatures", "cpufeatures",
@@ -1920,9 +1913,9 @@ dependencies = [
[[package]] [[package]]
name = "syn" name = "syn"
version = "1.0.106" version = "1.0.103"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09ee3a69cd2c7e06684677e5629b3878b253af05e4714964204279c6bc02cf0b" checksum = "a864042229133ada95abf3b54fdc62ef5ccabe9515b64717bcb9a1919e59445d"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@@ -1960,18 +1953,18 @@ dependencies = [
[[package]] [[package]]
name = "thiserror" name = "thiserror"
version = "1.0.38" version = "1.0.37"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a9cd18aa97d5c45c6603caea1da6628790b37f7a34b6ca89522331c5180fed0" checksum = "10deb33631e3c9018b9baf9dcbbc4f737320d2b576bac10f6aefa048fa407e3e"
dependencies = [ dependencies = [
"thiserror-impl", "thiserror-impl",
] ]
[[package]] [[package]]
name = "thiserror-impl" name = "thiserror-impl"
version = "1.0.38" version = "1.0.37"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fb327af4685e4d03fa8cbcf1716380da910eeb2bb8be417e7f9fd3fb164f36f" checksum = "982d17546b47146b28f7c22e3d08465f6b8903d0ea13c1660d9d84a6e7adcdbb"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@@ -2004,9 +1997,9 @@ checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
[[package]] [[package]]
name = "tokio" name = "tokio"
version = "1.23.0" version = "1.21.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eab6d665857cc6ca78d6e80303a02cea7a7851e85dfbd77cbdc09bd129f1ef46" checksum = "a9e03c497dc955702ba729190dc4aac6f2a0ce97f913e5b1b5912fc5039d9099"
dependencies = [ dependencies = [
"autocfg 1.1.0", "autocfg 1.1.0",
"bytes", "bytes",
@@ -2020,7 +2013,7 @@ dependencies = [
"socket2", "socket2",
"tokio-macros", "tokio-macros",
"tracing", "tracing",
"windows-sys 0.42.0", "winapi",
] ]
[[package]] [[package]]
@@ -2035,9 +2028,9 @@ dependencies = [
[[package]] [[package]]
name = "tokio-macros" name = "tokio-macros"
version = "1.8.2" version = "1.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d266c00fde287f55d3f1c3e96c500c362a2b8c695076ec180f27918820bc6df8" checksum = "9724f9a975fb987ef7a3cd9be0350edcbe130698af5b8f7a631e23d42d052484"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@@ -2093,18 +2086,18 @@ dependencies = [
[[package]] [[package]]
name = "toml" name = "toml"
version = "0.5.10" version = "0.5.9"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1333c76748e868a4d9d1017b5ab53171dfd095f70c712fdb4653a406547f598f" checksum = "8d82e1a7758622a465f8cee077614c73484dac5b836c02ff6a40d5d1010324d7"
dependencies = [ dependencies = [
"serde", "serde",
] ]
[[package]] [[package]]
name = "tonic" name = "tonic"
version = "0.8.3" version = "0.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f219fad3b929bef19b1f86fbc0358d35daed8f2cac972037ac0dc10bbb8d5fb" checksum = "55b9af819e54b8f33d453655bef9b9acc171568fb49523078d0cc4e7484200ec"
dependencies = [ dependencies = [
"async-stream", "async-stream",
"async-trait", "async-trait",
@@ -2154,9 +2147,9 @@ dependencies = [
[[package]] [[package]]
name = "tower-http" name = "tower-http"
version = "0.3.5" version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f873044bf02dd1e8239e9c1293ea39dad76dc594ec16185d0a1bf31d8dc8d858" checksum = "3c530c8675c1dbf98facee631536fa116b5fb6382d7dd6dc1b118d970eafe3ba"
dependencies = [ dependencies = [
"bitflags", "bitflags",
"bytes", "bytes",
@@ -2312,9 +2305,9 @@ dependencies = [
[[package]] [[package]]
name = "typenum" name = "typenum"
version = "1.16.0" version = "1.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "497961ef93d974e23eb6f433eb5fe1b7930b659f06d12dec6fc44a8f554c0bba" checksum = "dcf81ac59edc17cc8697ff311e8f5ef2d99fcbd9817b34cec66f90b6c3dfd987"
[[package]] [[package]]
name = "ucd-trie" name = "ucd-trie"
@@ -2330,9 +2323,9 @@ checksum = "099b7128301d285f79ddd55b9a83d5e6b9e97c92e0ea0daebee7263e932de992"
[[package]] [[package]]
name = "unicode-ident" name = "unicode-ident"
version = "1.0.6" version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "84a22b9f218b40614adcb3f4ff08b703773ad44fa9423e4e0d346d5db86e4ebc" checksum = "6ceab39d59e4c9499d4e5a8ee0e2735b891bb7308ac83dfb4e80cad195c9f6f3"
[[package]] [[package]]
name = "unicode-normalization" name = "unicode-normalization"
@@ -2374,9 +2367,9 @@ checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
[[package]] [[package]]
name = "uuid" name = "uuid"
version = "1.2.2" version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "422ee0de9031b5b948b97a8fc04e3aa35230001a722ddd27943e0be31564ce4c" checksum = "feb41e78f93363bb2df8b0e86a2ca30eed7806ea16ea0c790d757cf93f79be83"
dependencies = [ dependencies = [
"getrandom", "getrandom",
] ]
+1 -9
View File
@@ -1,15 +1,7 @@
[package] [package]
name = "nostr-rs-relay" name = "nostr-rs-relay"
version = "0.7.12" version = "0.7.1"
edition = "2021" edition = "2021"
authors = ["Greg Heartsfield <scsibug@imap.cc>"]
description = "A relay implementation for the Nostr protocol"
readme = "README.md"
homepage = "https://sr.ht/~gheartsfield/nostr-rs-relay/"
repository = "https://git.sr.ht/~gheartsfield/nostr-rs-relay"
license = "MIT"
keywords = ["nostr", "server"]
categories = ["network-programming", "web-programming"]
[dependencies] [dependencies]
tracing = "0.1.36" tracing = "0.1.36"
+4 -9
View File
@@ -1,23 +1,18 @@
FROM docker.io/library/rust:1.66.0@sha256:359949280cebefe93ccb33089fe25111a3aadfe99eac4b6cbe8ec3e1b571dacb as builder FROM docker.io/library/rust:1.65.0@sha256:1bca14676a365d0ed37a1e2a1da86c2bcf883fdf6e6886469434763d94d4afd5 as builder
RUN USER=root cargo install cargo-auditable
RUN USER=root cargo new --bin nostr-rs-relay RUN USER=root cargo new --bin nostr-rs-relay
WORKDIR ./nostr-rs-relay WORKDIR ./nostr-rs-relay
COPY ./Cargo.toml ./Cargo.toml COPY ./Cargo.toml ./Cargo.toml
COPY ./Cargo.lock ./Cargo.lock COPY ./Cargo.lock ./Cargo.lock
# build dependencies only (caching) RUN cargo build --release
RUN cargo auditable build --release --locked
# get rid of starter project code
RUN rm src/*.rs RUN rm src/*.rs
# copy project source code
COPY ./src ./src COPY ./src ./src
# build auditable release using locked deps
RUN rm ./target/release/deps/nostr*relay* RUN rm ./target/release/deps/nostr*relay*
RUN cargo auditable build --release --locked RUN cargo build --release
FROM docker.io/library/debian:bullseye-20221205-slim@sha256:25f10b4f1ded5341a3ca0a30290ff3cd5639415f0c5a2222d5e7d5dd72952aa1 FROM docker.io/library/debian:bullseye-20221024-slim@sha256:76cdda8fe5eb597ef5e712e4c9a9f5f1fb119e69f353daaa7bd6d0f6e66e541d
ARG APP=/usr/src/app ARG APP=/usr/src/app
ARG APP_DATA=/usr/src/app/db ARG APP_DATA=/usr/src/app/db
+9 -27
View File
@@ -1,8 +1,8 @@
# [nostr-rs-relay](https://git.sr.ht/~gheartsfield/nostr-rs-relay) # [nostr-rs-relay](https://git.sr.ht/~gheartsfield/nostr-rs-relay)
This is a [nostr](https://github.com/nostr-protocol/nostr) relay, This is a [nostr](https://github.com/nostr-protocol/nostr) relay, written in
written in Rust. It currently supports the entire relay protocol, and Rust. It currently supports the entire relay protocol, and has a
persists data with SQLite. SQLite persistence layer.
The project master repository is available on The project master repository is available on
[sourcehut](https://sr.ht/~gheartsfield/nostr-rs-relay/), and is [sourcehut](https://sr.ht/~gheartsfield/nostr-rs-relay/), and is
@@ -26,7 +26,6 @@ mirrored on [GitHub](https://github.com/scsibug/nostr-rs-relay).
- [x] NIP-12: [Generic Tag Queries](https://github.com/nostr-protocol/nips/blob/master/12.md) - [x] NIP-12: [Generic Tag Queries](https://github.com/nostr-protocol/nips/blob/master/12.md)
- [x] NIP-15: [End of Stored Events Notice](https://github.com/nostr-protocol/nips/blob/master/15.md) - [x] NIP-15: [End of Stored Events Notice](https://github.com/nostr-protocol/nips/blob/master/15.md)
- [x] NIP-16: [Event Treatment](https://github.com/nostr-protocol/nips/blob/master/16.md) - [x] NIP-16: [Event Treatment](https://github.com/nostr-protocol/nips/blob/master/16.md)
- [x] NIP-20: [Command Results](https://github.com/nostr-protocol/nips/blob/master/20.md)
- [x] NIP-22: [Event `created_at` limits](https://github.com/nostr-protocol/nips/blob/master/22.md) (_future-dated events only_) - [x] NIP-22: [Event `created_at` limits](https://github.com/nostr-protocol/nips/blob/master/22.md) (_future-dated events only_)
- [x] NIP-26: [Event Delegation](https://github.com/nostr-protocol/nips/blob/master/26.md) - [x] NIP-26: [Event Delegation](https://github.com/nostr-protocol/nips/blob/master/26.md)
@@ -37,32 +36,15 @@ application. Use a bind mount to store the SQLite database outside of
the container image, and map the container's 8080 port to a host port the container image, and map the container's 8080 port to a host port
(7000 in the example below). (7000 in the example below).
The examples below start a rootless podman container, mapping a local
data directory and config file.
```console ```console
$ podman build -t nostr-rs-relay . $ docker build -t nostr-rs-relay .
$ mkdir data $ docker run -it -p 7000:8080 \
--mount src=$(pwd)/data,target=/usr/src/app/db,type=bind nostr-rs-relay
$ podman unshare chown 100:100 data [2021-12-31T19:58:31Z INFO nostr_rs_relay] listening on: 0.0.0.0:8080
[2021-12-31T19:58:31Z INFO nostr_rs_relay::db] opened database "/usr/src/app/db/nostr.db" for writing
$ podman run -it --rm -p 7000:8080 \ [2021-12-31T19:58:31Z INFO nostr_rs_relay::db] DB version = 2
--user=100:100 \
-v $(pwd)/data:/usr/src/app/db:Z \
-v $(pwd)/config.toml:/usr/src/app/config.toml:ro,Z \
--name nostr-relay nostr-rs-relay:latest
Nov 19 15:31:15.013 INFO nostr_rs_relay: Starting up from main
Nov 19 15:31:15.017 INFO nostr_rs_relay::server: listening on: 0.0.0.0:8080
Nov 19 15:31:15.019 INFO nostr_rs_relay::server: db writer created
Nov 19 15:31:15.019 INFO nostr_rs_relay::server: control message listener started
Nov 19 15:31:15.019 INFO nostr_rs_relay::db: Built a connection pool "event writer" (min=1, max=4)
Nov 19 15:31:15.019 INFO nostr_rs_relay::db: opened database "/usr/src/app/db/nostr.db" for writing
Nov 19 15:31:15.019 INFO nostr_rs_relay::schema: DB version = 0
Nov 19 15:31:15.054 INFO nostr_rs_relay::schema: database pragma/schema initialized to v7, and ready
Nov 19 15:31:15.054 INFO nostr_rs_relay::schema: All migration scripts completed successfully. Welcome to v7.
Nov 19 15:31:15.521 INFO nostr_rs_relay::db: Built a connection pool "client query" (min=4, max=128)
``` ```
Use a `nostr` client such as Use a `nostr` client such as
+2 -18
View File
@@ -62,24 +62,8 @@ reject_future_seconds = 1800
[limits] [limits]
# Limit events created per second, averaged over one minute. Must be # Limit events created per second, averaged over one minute. Must be
# an integer. If not set (or set to 0), defaults to unlimited. Note: # an integer. If not set (or set to 0), defaults to unlimited.
# this is for the server as a whole, not per-connection. #messages_per_sec = 0
# messages_per_sec = 0
# Limit client subscriptions created per second, averaged over one
# minute. Must be an integer. If not set (or set to 0), defaults to
# unlimited.
#subscriptions_per_min = 0
# UNIMPLEMENTED...
# Limit how many concurrent database connections a client can have.
# This prevents a single client from starting too many expensive
# database queries. Must be an integer. If not set (or set to 0),
# defaults to unlimited (subject to subscription limits).
#db_conns_per_client = 0
# Limit blocking threads used for database connections. Defaults to 16.
#max_blocking_threads = 16
# Limit the maximum size of an EVENT message. Defaults to 128 KB. # Limit the maximum size of an EVENT message. Defaults to 128 KB.
# Set to 0 for unlimited. # Set to 0 for unlimited.
-6
View File
@@ -52,9 +52,6 @@ pub struct Retention {
#[allow(unused)] #[allow(unused)]
pub struct Limits { pub struct Limits {
pub messages_per_sec: Option<u32>, // Artificially slow down event writing to limit disk consumption (averaged over 1 minute) pub messages_per_sec: Option<u32>, // Artificially slow down event writing to limit disk consumption (averaged over 1 minute)
pub subscriptions_per_min: Option<u32>, // Artificially slow down request (db query) creation to prevent abuse (averaged over 1 minute)
pub db_conns_per_client: Option<u32>, // How many concurrent database queries (not subscriptions) may a client have?
pub max_blocking_threads: usize,
pub max_event_bytes: Option<usize>, // Maximum size of an EVENT message pub max_event_bytes: Option<usize>, // Maximum size of an EVENT message
pub max_ws_message_bytes: Option<usize>, pub max_ws_message_bytes: Option<usize>,
pub max_ws_frame_bytes: Option<usize>, pub max_ws_frame_bytes: Option<usize>,
@@ -217,9 +214,6 @@ impl Default for Settings {
}, },
limits: Limits { limits: Limits {
messages_per_sec: None, messages_per_sec: None,
subscriptions_per_min: None,
db_conns_per_client: None,
max_blocking_threads: 16,
max_event_bytes: Some(2 << 17), // 128K max_event_bytes: Some(2 << 17), // 128K
max_ws_message_bytes: Some(2 << 17), // 128K max_ws_message_bytes: Some(2 << 17), // 128K
max_ws_frame_bytes: Some(2 << 17), // 128K max_ws_frame_bytes: Some(2 << 17), // 128K
+5 -15
View File
@@ -46,11 +46,6 @@ impl ClientConn {
&self.subscriptions &self.subscriptions
} }
/// Check if the given subscription already exists
pub fn has_subscription(&self, sub: &Subscription) -> bool {
self.subscriptions.values().any(|x| x == sub)
}
/// Get a short prefix of the client's unique identifier, suitable /// Get a short prefix of the client's unique identifier, suitable
/// for logging. /// for logging.
#[must_use] #[must_use]
@@ -83,12 +78,8 @@ impl ClientConn {
// check if an existing subscription exists, and replace if so // check if an existing subscription exists, and replace if so
if self.subscriptions.contains_key(&k) { if self.subscriptions.contains_key(&k) {
self.subscriptions.remove(&k); self.subscriptions.remove(&k);
self.subscriptions.insert(k, s.clone()); self.subscriptions.insert(k, s);
debug!( debug!("replaced existing subscription");
"replaced existing subscription (cid: {}, sub: {:?})",
self.get_client_prefix(),
s.get_id()
);
return Ok(()); return Ok(());
} }
@@ -99,9 +90,8 @@ impl ClientConn {
// add subscription // add subscription
self.subscriptions.insert(k, s); self.subscriptions.insert(k, s);
debug!( debug!(
"registered new subscription, currently have {} active subs (cid: {})", "registered new subscription, currently have {} active subs",
self.subscriptions.len(), self.subscriptions.len()
self.get_client_prefix(),
); );
Ok(()) Ok(())
} }
@@ -111,7 +101,7 @@ impl ClientConn {
// TODO: return notice if subscription did not exist. // TODO: return notice if subscription did not exist.
self.subscriptions.remove(&c.id); self.subscriptions.remove(&c.id);
debug!( debug!(
"removed subscription, currently have {} active subs (cid: {})", "removed subscription, currently have {} active subs (cid={})",
self.subscriptions.len(), self.subscriptions.len(),
self.get_client_prefix(), self.get_client_prefix(),
); );
+36 -135
View File
@@ -38,8 +38,6 @@ pub struct SubmittedEvent {
/// Database file /// Database file
pub const DB_FILE: &str = "nostr.db"; pub const DB_FILE: &str = "nostr.db";
/// How many persisted events before optimization is triggered
pub const EVENT_COUNT_OPTIMIZE_TRIGGER: usize = 500;
/// Build a database connection pool. /// Build a database connection pool.
/// # Panics /// # Panics
@@ -78,7 +76,6 @@ pub fn build_pool(
.test_on_check_out(true) // no noticeable performance hit .test_on_check_out(true) // no noticeable performance hit
.min_idle(Some(min_size)) .min_idle(Some(min_size))
.max_size(max_size) .max_size(max_size)
.max_lifetime(Some(Duration::from_secs(60)))
.build(manager) .build(manager)
.unwrap(); .unwrap();
info!( info!(
@@ -88,12 +85,6 @@ pub fn build_pool(
pool pool
} }
/// Perform normal maintenance
pub fn optimize_db(conn: &mut PooledConnection) -> Result<()> {
conn.execute_batch("PRAGMA optimize;")?;
Ok(())
}
/// Spawn a database writer that persists events to the SQLite store. /// Spawn a database writer that persists events to the SQLite store.
pub async fn db_writer( pub async fn db_writer(
settings: Settings, settings: Settings,
@@ -116,7 +107,7 @@ pub async fn db_writer(
&settings, &settings,
OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE, OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE,
1, 1,
2, 4,
false, false,
); );
if settings.database.in_memory { if settings.database.in_memory {
@@ -133,8 +124,6 @@ pub async fn db_writer(
let rps_setting = settings.limits.messages_per_sec; let rps_setting = settings.limits.messages_per_sec;
let mut most_recent_rate_limit = Instant::now(); let mut most_recent_rate_limit = Instant::now();
let mut lim_opt = None; let mut lim_opt = None;
// Keep rough track of events so we can run optimize eventually.
let mut optimize_counter: usize = 0;
let clock = governor::clock::QuantaClock::default(); let clock = governor::clock::QuantaClock::default();
if let Some(rps) = rps_setting { if let Some(rps) = rps_setting {
if rps > 0 { if rps > 0 {
@@ -170,9 +159,8 @@ pub async fn db_writer(
event.get_event_id_prefix() event.get_event_id_prefix()
); );
notice_tx notice_tx
.try_send(Notice::blocked( .try_send(Notice::message(
event.id, "pubkey is not allowed to publish to this relay".to_owned(),
"pubkey is not allowed to publish to this relay",
)) ))
.ok(); .ok();
continue; continue;
@@ -204,9 +192,9 @@ pub async fn db_writer(
event.get_author_prefix() event.get_author_prefix()
); );
notice_tx notice_tx
.try_send(Notice::blocked( .try_send(Notice::message(
event.id, "NIP-05 verification is no longer valid (expired/wrong domain)"
"NIP-05 verification is no longer valid (expired/wrong domain)", .to_owned(),
)) ))
.ok(); .ok();
continue; continue;
@@ -218,9 +206,8 @@ pub async fn db_writer(
event.get_author_prefix() event.get_author_prefix()
); );
notice_tx notice_tx
.try_send(Notice::blocked( .try_send(Notice::message(
event.id, "NIP-05 verification needed to publish events".to_owned(),
"NIP-05 verification needed to publish events",
)) ))
.ok(); .ok();
continue; continue;
@@ -263,17 +250,11 @@ pub async fn db_writer(
} }
Err(err) => { Err(err) => {
warn!("event insert failed: {:?}", err); warn!("event insert failed: {:?}", err);
let msg = "relay experienced an error trying to publish the latest event"; let msg =
notice_tx.try_send(Notice::error(event.id, msg)).ok(); "relay experienced an error trying to publish the latest event".into();
notice_tx.try_send(Notice::err_msg(msg, event.id)).ok();
} }
} }
// Use this as a trigger to do optimization
optimize_counter += 1;
if optimize_counter > EVENT_COUNT_OPTIMIZE_TRIGGER {
info!("running database optimizer");
optimize_counter = 0;
optimize_db(&mut pool.get()?).ok();
}
} }
// use rate limit, if defined, and if an event was actually written. // use rate limit, if defined, and if an event was actually written.
@@ -446,13 +427,14 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
// if the filter is malformed, don't return anything. // if the filter is malformed, don't return anything.
if f.force_no_match { if f.force_no_match {
let empty_query = "SELECT e.content, e.created_at FROM event e WHERE 1=0".to_owned(); let empty_query =
"SELECT DISTINCT(e.content), e.created_at FROM event e WHERE 1=0".to_owned();
// query parameters for SQLite // query parameters for SQLite
let empty_params: Vec<Box<dyn ToSql>> = vec![]; let empty_params: Vec<Box<dyn ToSql>> = vec![];
return (empty_query, empty_params); return (empty_query, empty_params);
} }
let mut query = "SELECT e.content, e.created_at FROM event e".to_owned(); let mut query = "SELECT DISTINCT(e.content), e.created_at FROM event e ".to_owned();
// query parameters for SQLite // query parameters for SQLite
let mut params: Vec<Box<dyn ToSql>> = vec![]; let mut params: Vec<Box<dyn ToSql>> = vec![];
@@ -488,14 +470,8 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
} }
} }
} }
if !authvec.is_empty() { let authors_clause = format!("({})", auth_searches.join(" OR "));
let authors_clause = format!("({})", auth_searches.join(" OR ")); filter_components.push(authors_clause);
filter_components.push(authors_clause);
} else {
// if the authors list was empty, we should never return
// any results.
filter_components.push("false".to_owned());
}
} }
// Query for Kind // Query for Kind
if let Some(ks) = &f.kinds { if let Some(ks) = &f.kinds {
@@ -528,14 +504,8 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
} }
} }
} }
if !idvec.is_empty() { let id_clause = format!("({})", id_searches.join(" OR "));
let id_clause = format!("({})", id_searches.join(" OR ")); filter_components.push(id_clause);
filter_components.push(id_clause);
} else {
// if the ids list was empty, we should never return
// any results.
filter_components.push("false".to_owned());
}
} }
// Query for tags // Query for tags
if let Some(map) = &f.tags { if let Some(map) = &f.tags {
@@ -608,21 +578,13 @@ fn query_from_sub(sub: &Subscription) -> (String, Vec<Box<dyn ToSql>>) {
// encapsulate subqueries into select statements // encapsulate subqueries into select statements
let subqueries_selects: Vec<String> = subqueries let subqueries_selects: Vec<String> = subqueries
.iter() .iter()
.map(|s| format!("SELECT distinct content, created_at FROM ({})", s)) .map(|s| format!("SELECT content, created_at FROM ({})", s))
.collect(); .collect();
let query: String = subqueries_selects.join(" UNION "); let query: String = subqueries_selects.join(" UNION ");
trace!("final query string: {}", query);
(query, params) (query, params)
} }
fn log_pool_stats(pool: &SqlitePool) {
let state: r2d2::State = pool.state();
let in_use_cxns = state.connections - state.idle_connections;
debug!(
"DB pool usage (in_use: {}, available: {})",
in_use_cxns, state.connections
);
}
/// Perform a database query using a subscription. /// Perform a database query using a subscription.
/// ///
/// The [`Subscription`] is converted into a SQL query. Each result /// The [`Subscription`] is converted into a SQL query. Each result
@@ -636,105 +598,45 @@ pub async fn db_query(
query_tx: tokio::sync::mpsc::Sender<QueryResult>, query_tx: tokio::sync::mpsc::Sender<QueryResult>,
mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>, mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>,
) { ) {
let pre_spawn_start = Instant::now();
task::spawn_blocking(move || { task::spawn_blocking(move || {
let db_queue_time = pre_spawn_start.elapsed(); trace!("going to query for: {:?}", sub);
// report queuing time if it is slow
if db_queue_time > Duration::from_secs(1) {
debug!(
"(slow) DB query queued for {:?} (cid: {}, sub: {:?})",
db_queue_time, client_id, sub.id
);
}
let start = Instant::now();
let mut row_count: usize = 0; let mut row_count: usize = 0;
let start = Instant::now();
// generate SQL query // generate SQL query
let (q, p) = query_from_sub(&sub); let (q, p) = query_from_sub(&sub);
debug!("SQL generated in {:?}", start.elapsed()); trace!("SQL generated in {:?}", start.elapsed());
// show pool stats // show pool stats
log_pool_stats(&pool); debug!("DB pool stats: {:?}", pool.state());
// cutoff for displaying slow queries
let slow_cutoff = Duration::from_millis(2000);
// any client that doesn't cause us to generate new rows in 5
// seconds gets dropped.
let abort_cutoff = Duration::from_secs(5);
let start = Instant::now(); let start = Instant::now();
let mut slow_first_event;
let mut last_successful_send = Instant::now();
if let Ok(conn) = pool.get() { if let Ok(conn) = pool.get() {
// execute the query. Don't cache, since queries vary so much. // execute the query. Don't cache, since queries vary so much.
let mut stmt = conn.prepare(&q)?; let mut stmt = conn.prepare(&q)?;
let mut event_rows = stmt.query(rusqlite::params_from_iter(p))?; let mut event_rows = stmt.query(rusqlite::params_from_iter(p))?;
let mut first_result = true; let mut first_result = true;
while let Some(row) = event_rows.next()? { while let Some(row) = event_rows.next()? {
let first_event_elapsed = start.elapsed();
slow_first_event = first_event_elapsed >= slow_cutoff;
if first_result { if first_result {
debug!( debug!(
"first result in {:?} (cid: {}, sub: {:?})", "time to first result: {:?} (cid={}, sub={:?})",
first_event_elapsed, client_id, sub.id start.elapsed(),
client_id,
sub.id
); );
first_result = false; first_result = false;
} }
// logging for slow queries; show sub and SQL. // check if this is still active
// to reduce logging; only show 1/16th of clients (leading 0) // TODO: check every N rows
if slow_first_event && client_id.starts_with("00") { if abandon_query_rx.try_recv().is_ok() {
debug!( debug!("query aborted (sub={:?})", sub.id);
"query req (slow): {:?} (cid: {}, sub: {:?})",
sub, client_id, sub.id
);
debug!(
"query string (slow): {} (cid: {}, sub: {:?})",
q, client_id, sub.id
);
} else {
trace!(
"query req: {:?} (cid: {}, sub: {:?})",
sub,
client_id,
sub.id
);
trace!(
"query string: {} (cid: {}, sub: {:?})",
q,
client_id,
sub.id
);
}
// check if this is still active; every 100 rows
if row_count % 100 == 0 && abandon_query_rx.try_recv().is_ok() {
debug!("query aborted (cid: {}, sub: {:?})", client_id, sub.id);
return Ok(()); return Ok(());
} }
row_count += 1; row_count += 1;
let event_json = row.get(0)?; let event_json = row.get(0)?;
loop {
if query_tx.capacity() != 0 {
// we have capacity to add another item
break;
} else {
// the queue is full
trace!("db reader thread is stalled");
if last_successful_send + abort_cutoff < Instant::now() {
// the queue has been full for too long, abort
info!("aborting database query due to slow client");
let ok: Result<()> = Ok(());
return ok;
}
// give the queue a chance to clear before trying again
thread::sleep(Duration::from_millis(100));
}
}
// TODO: we could use try_send, but we'd have to juggle
// getting the query result back as part of the error
// result.
query_tx query_tx
.blocking_send(QueryResult { .blocking_send(QueryResult {
sub_id: sub.get_id(), sub_id: sub.get_id(),
event: event_json, event: event_json,
}) })
.ok(); .ok();
last_successful_send = Instant::now();
} }
query_tx query_tx
.blocking_send(QueryResult { .blocking_send(QueryResult {
@@ -743,12 +645,11 @@ pub async fn db_query(
}) })
.ok(); .ok();
debug!( debug!(
"query completed in {:?} (cid: {}, sub: {:?}, db_time: {:?}, rows: {})", "query completed ({} rows) in {:?} (cid={}, sub={:?})",
pre_spawn_start.elapsed(), row_count,
client_id,
sub.id,
start.elapsed(), start.elapsed(),
row_count client_id,
sub.id
); );
} else { } else {
warn!("Could not get a database connection for querying"); warn!("Could not get a database connection for querying");
+3 -9
View File
@@ -17,16 +17,10 @@ pub enum Error {
ConnWriteError, ConnWriteError,
#[error("EVENT parse failed")] #[error("EVENT parse failed")]
EventParseFailed, EventParseFailed,
#[error("CLOSE message parse failed")] #[error("ClOSE message parse failed")]
CloseParseFailed, CloseParseFailed,
#[error("Event invalid signature")] #[error("Event validation failed")]
EventInvalidSignature, EventInvalid,
#[error("Event invalid id")]
EventInvalidId,
#[error("Event malformed pubkey")]
EventMalformedPubkey,
#[error("Event could not canonicalize")]
EventCouldNotCanonicalize,
#[error("Event too large")] #[error("Event too large")]
EventMaxLengthError(usize), EventMaxLengthError(usize),
#[error("Subscription identifier max length exceeded")] #[error("Subscription identifier max length exceeded")]
+14 -20
View File
@@ -27,12 +27,6 @@ pub struct EventCmd {
event: Event, event: Event,
} }
impl EventCmd {
pub fn event_id(&self) -> &str {
&self.event.id
}
}
/// Parsed nostr event. /// Parsed nostr event.
#[derive(Serialize, Deserialize, PartialEq, Eq, Debug, Clone)] #[derive(Serialize, Deserialize, PartialEq, Eq, Debug, Clone)]
pub struct Event { pub struct Event {
@@ -89,13 +83,13 @@ impl From<EventCmd> for Result<Event> {
// ensure command is correct // ensure command is correct
if ec.cmd != "EVENT" { if ec.cmd != "EVENT" {
Err(CommandUnknownError) Err(CommandUnknownError)
} else if ec.event.is_valid() {
let mut e = ec.event;
e.build_index();
e.update_delegation();
Ok(e)
} else { } else {
ec.event.validate().map(|_| { Err(EventInvalid)
let mut e = ec.event;
e.build_index();
e.update_delegation();
e
})
} }
} }
} }
@@ -226,7 +220,7 @@ impl Event {
} }
/// Check if this event has a valid signature. /// Check if this event has a valid signature.
fn validate(&self) -> Result<()> { fn is_valid(&self) -> bool {
// TODO: return a Result with a reason for invalid events // TODO: return a Result with a reason for invalid events
// validation is performed by: // validation is performed by:
// * parsing JSON string into event fields // * parsing JSON string into event fields
@@ -235,8 +229,8 @@ impl Event {
// * serialize with no spaces/newlines // * serialize with no spaces/newlines
let c_opt = self.to_canonical(); let c_opt = self.to_canonical();
if c_opt.is_none() { if c_opt.is_none() {
debug!("could not canonicalize"); debug!("event could not be canonicalized");
return Err(EventCouldNotCanonicalize); return false;
} }
let c = c_opt.unwrap(); let c = c_opt.unwrap();
// * compute the sha256sum. // * compute the sha256sum.
@@ -245,21 +239,21 @@ impl Event {
// * ensure the id matches the computed sha256sum. // * ensure the id matches the computed sha256sum.
if self.id != hex_digest { if self.id != hex_digest {
debug!("event id does not match digest"); debug!("event id does not match digest");
return Err(EventInvalidId); return false;
} }
// * validate the message digest (sig) using the pubkey & computed sha256 message hash. // * validate the message digest (sig) using the pubkey & computed sha256 message hash.
let sig = schnorr::Signature::from_str(&self.sig).unwrap(); let sig = schnorr::Signature::from_str(&self.sig).unwrap();
if let Ok(msg) = secp256k1::Message::from_slice(digest.as_ref()) { if let Ok(msg) = secp256k1::Message::from_slice(digest.as_ref()) {
if let Ok(pubkey) = XOnlyPublicKey::from_str(&self.pubkey) { if let Ok(pubkey) = XOnlyPublicKey::from_str(&self.pubkey) {
SECP.verify_schnorr(&sig, &msg, &pubkey) let verify = SECP.verify_schnorr(&sig, &msg, &pubkey);
.map_err(|_| EventInvalidSignature) matches!(verify, Ok(()))
} else { } else {
debug!("client sent malformed pubkey"); debug!("client sent malformed pubkey");
Err(EventMalformedPubkey) false
} }
} else { } else {
info!("error converting digest to secp256k1 message"); info!("error converting digest to secp256k1 message");
Err(EventInvalidSignature) false
} }
} }
+1 -1
View File
@@ -3,7 +3,7 @@ use crate::utils::is_hex;
use hex; use hex;
/// Types of hexadecimal queries. /// Types of hexadecimal queries.
#[derive(PartialEq, Eq, PartialOrd, Ord, Debug, Clone)] #[derive(PartialEq, Eq, Debug, Clone)]
pub enum HexSearch { pub enum HexSearch {
// when no range is needed, exact 32-byte // when no range is needed, exact 32-byte
Exact(Vec<u8>), Exact(Vec<u8>),
+1 -1
View File
@@ -35,7 +35,7 @@ impl From<config::Info> for RelayInfo {
description: i.description, description: i.description,
pubkey: i.pubkey, pubkey: i.pubkey,
contact: i.contact, contact: i.contact,
supported_nips: Some(vec![1, 2, 9, 11, 12, 15, 16, 20, 22, 26]), supported_nips: Some(vec![1, 2, 9, 11, 12, 15, 16, 22, 26]),
software: Some("https://git.sr.ht/~gheartsfield/nostr-rs-relay".to_owned()), software: Some("https://git.sr.ht/~gheartsfield/nostr-rs-relay".to_owned()),
version: CARGO_PKG_VERSION.map(|x| x.to_owned()), version: CARGO_PKG_VERSION.map(|x| x.to_owned()),
} }
+20 -58
View File
@@ -1,15 +1,13 @@
use crate::error;
pub enum EventResultStatus { pub enum EventResultStatus {
Saved, Saved,
Duplicate, Duplicate,
Invalid, Error(String),
Blocked,
RateLimited,
Error,
} }
pub struct EventResult { pub struct EventResult {
pub id: String, pub id: String,
pub msg: String,
pub status: EventResultStatus, pub status: EventResultStatus,
} }
@@ -18,69 +16,33 @@ pub enum Notice {
EventResult(EventResult), EventResult(EventResult),
} }
impl EventResultStatus {
pub fn to_bool(&self) -> bool {
match self {
Self::Saved => true,
Self::Duplicate => true,
Self::Invalid => false,
Self::Blocked => false,
Self::RateLimited => false,
Self::Error => false,
}
}
pub fn prefix(&self) -> &'static str {
match self {
Self::Saved => "saved",
Self::Duplicate => "duplicate",
Self::Invalid => "invalid",
Self::Blocked => "blocked",
Self::RateLimited => "rate-limited",
Self::Error => "error",
}
}
}
impl Notice { impl Notice {
//pub fn err(err: error::Error, id: String) -> Notice { pub fn err(err: error::Error, id: String) -> Notice {
// Notice::err_msg(format!("{}", err), id) Notice::err_msg(format!("{}", err), id)
//} }
pub fn message(msg: String) -> Notice { pub fn message(msg: String) -> Notice {
Notice::Message(msg) Notice::Message(msg)
} }
fn prefixed(id: String, msg: &str, status: EventResultStatus) -> Notice {
let msg = format!("{}: {}", status.prefix(), msg);
Notice::EventResult(EventResult { id, msg, status })
}
pub fn invalid(id: String, msg: &str) -> Notice {
Notice::prefixed(id, msg, EventResultStatus::Invalid)
}
pub fn blocked(id: String, msg: &str) -> Notice {
Notice::prefixed(id, msg, EventResultStatus::Blocked)
}
pub fn rate_limited(id: String, msg: &str) -> Notice {
Notice::prefixed(id, msg, EventResultStatus::RateLimited)
}
pub fn duplicate(id: String) -> Notice {
Notice::prefixed(id, "", EventResultStatus::Duplicate)
}
pub fn error(id: String, msg: &str) -> Notice {
Notice::prefixed(id, msg, EventResultStatus::Error)
}
pub fn saved(id: String) -> Notice { pub fn saved(id: String) -> Notice {
Notice::EventResult(EventResult { Notice::EventResult(EventResult {
id, id,
msg: "".into(),
status: EventResultStatus::Saved, status: EventResultStatus::Saved,
}) })
} }
pub fn duplicate(id: String) -> Notice {
Notice::EventResult(EventResult {
id,
status: EventResultStatus::Duplicate,
})
}
pub fn err_msg(msg: String, id: String) -> Notice {
Notice::EventResult(EventResult {
id,
status: EventResultStatus::Error(msg),
})
}
} }
+4 -100
View File
@@ -16,11 +16,11 @@ pub const STARTUP_SQL: &str = r##"
PRAGMA main.synchronous=NORMAL; PRAGMA main.synchronous=NORMAL;
PRAGMA foreign_keys = ON; PRAGMA foreign_keys = ON;
PRAGMA journal_size_limit=32768; PRAGMA journal_size_limit=32768;
pragma mmap_size = 17179869184; -- cap mmap at 16GB pragma mmap_size = 536870912; -- 512MB of mmap
"##; "##;
/// Latest database version /// Latest database version
pub const DB_VERSION: usize = 11; pub const DB_VERSION: usize = 7;
/// Schema definition /// Schema definition
const INIT_SQL: &str = formatcp!( const INIT_SQL: &str = formatcp!(
@@ -48,10 +48,10 @@ content TEXT NOT NULL -- serialized json of event object
-- Event Indexes -- Event Indexes
CREATE UNIQUE INDEX IF NOT EXISTS event_hash_index ON event(event_hash); CREATE UNIQUE INDEX IF NOT EXISTS event_hash_index ON event(event_hash);
CREATE INDEX IF NOT EXISTS author_index ON event(author);
CREATE INDEX IF NOT EXISTS created_at_index ON event(created_at); CREATE INDEX IF NOT EXISTS created_at_index ON event(created_at);
CREATE INDEX IF NOT EXISTS author_index ON event(author);
CREATE INDEX IF NOT EXISTS delegated_by_index ON event(delegated_by); CREATE INDEX IF NOT EXISTS delegated_by_index ON event(delegated_by);
CREATE INDEX IF NOT EXISTS event_composite_index ON event(kind,created_at); CREATE INDEX IF NOT EXISTS kind_index ON event(kind);
-- Tag Table -- Tag Table
-- Tag values are stored as either a BLOB (if they come in as a -- Tag values are stored as either a BLOB (if they come in as a
@@ -67,8 +67,6 @@ FOREIGN KEY(event_id) REFERENCES event(id) ON UPDATE CASCADE ON DELETE CASCADE
); );
CREATE INDEX IF NOT EXISTS tag_val_index ON tag(value); CREATE INDEX IF NOT EXISTS tag_val_index ON tag(value);
CREATE INDEX IF NOT EXISTS tag_val_hex_index ON tag(value_hex); CREATE INDEX IF NOT EXISTS tag_val_hex_index ON tag(value_hex);
CREATE INDEX IF NOT EXISTS tag_composite_index ON tag(event_id,name,value_hex,value);
CREATE INDEX IF NOT EXISTS tag_name_eid_index ON tag(name,event_id,value_hex);
-- NIP-05 User Validation -- NIP-05 User Validation
CREATE TABLE IF NOT EXISTS user_verification ( CREATE TABLE IF NOT EXISTS user_verification (
@@ -159,19 +157,6 @@ pub fn upgrade_db(conn: &mut PooledConnection) -> Result<()> {
if curr_version == 6 { if curr_version == 6 {
curr_version = mig_6_to_7(conn)?; curr_version = mig_6_to_7(conn)?;
} }
if curr_version == 7 {
curr_version = mig_7_to_8(conn)?;
}
if curr_version == 8 {
curr_version = mig_8_to_9(conn)?;
}
if curr_version == 9 {
curr_version = mig_9_to_10(conn)?;
}
if curr_version == 10 {
curr_version = mig_10_to_11(conn)?;
}
if curr_version == DB_VERSION { if curr_version == DB_VERSION {
info!( info!(
"All migration scripts completed successfully. Welcome to v{}.", "All migration scripts completed successfully. Welcome to v{}.",
@@ -388,84 +373,3 @@ PRAGMA user_version = 7;
} }
Ok(7) Ok(7)
} }
fn mig_7_to_8(conn: &mut PooledConnection) -> Result<usize> {
info!("database schema needs update from 7->8");
// Remove redundant indexes, and add a better multi-column index.
let upgrade_sql = r##"
DROP INDEX IF EXISTS created_at_index;
DROP INDEX IF EXISTS kind_index;
CREATE INDEX IF NOT EXISTS event_composite_index ON event(kind,created_at);
PRAGMA user_version = 8;
"##;
match conn.execute_batch(upgrade_sql) {
Ok(()) => {
info!("database schema upgraded v7 -> v8");
}
Err(err) => {
error!("update failed: {}", err);
panic!("database could not be upgraded");
}
}
Ok(8)
}
fn mig_8_to_9(conn: &mut PooledConnection) -> Result<usize> {
info!("database schema needs update from 8->9");
// Those old indexes were actually helpful...
let upgrade_sql = r##"
CREATE INDEX IF NOT EXISTS created_at_index ON event(created_at);
CREATE INDEX IF NOT EXISTS event_composite_index ON event(kind,created_at);
PRAGMA user_version = 9;
"##;
match conn.execute_batch(upgrade_sql) {
Ok(()) => {
info!("database schema upgraded v8 -> v9");
}
Err(err) => {
error!("update failed: {}", err);
panic!("database could not be upgraded");
}
}
Ok(9)
}
fn mig_9_to_10(conn: &mut PooledConnection) -> Result<usize> {
info!("database schema needs update from 9->10");
// Those old indexes were actually helpful...
let upgrade_sql = r##"
CREATE INDEX IF NOT EXISTS tag_composite_index ON tag(event_id,name,value_hex,value);
PRAGMA user_version = 10;
"##;
match conn.execute_batch(upgrade_sql) {
Ok(()) => {
info!("database schema upgraded v9 -> v10");
}
Err(err) => {
error!("update failed: {}", err);
panic!("database could not be upgraded");
}
}
Ok(10)
}
fn mig_10_to_11(conn: &mut PooledConnection) -> Result<usize> {
info!("database schema needs update from 10->11");
// Those old indexes were actually helpful...
let upgrade_sql = r##"
CREATE INDEX IF NOT EXISTS tag_name_eid_index ON tag(name,event_id,value_hex);
reindex;
pragma optimize;
PRAGMA user_version = 11;
"##;
match conn.execute_batch(upgrade_sql) {
Ok(()) => {
info!("database schema upgraded v10 -> v11");
}
Err(err) => {
error!("update failed: {}", err);
panic!("database could not be upgraded");
}
}
Ok(11)
}
+44 -76
View File
@@ -10,11 +10,10 @@ use crate::event::Event;
use crate::event::EventCmd; use crate::event::EventCmd;
use crate::info::RelayInfo; use crate::info::RelayInfo;
use crate::nip05; use crate::nip05;
use crate::notice::Notice; use crate::notice::{EventResultStatus, Notice};
use crate::subscription::Subscription; use crate::subscription::Subscription;
use futures::SinkExt; use futures::SinkExt;
use futures::StreamExt; use futures::StreamExt;
use governor::{Jitter, Quota, RateLimiter};
use http::header::HeaderMap; use http::header::HeaderMap;
use hyper::header::ACCEPT; use hyper::header::ACCEPT;
use hyper::service::{make_service_fn, service_fn}; use hyper::service::{make_service_fn, service_fn};
@@ -245,14 +244,6 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
let rt = Builder::new_multi_thread() let rt = Builder::new_multi_thread()
.enable_all() .enable_all()
.thread_name("tokio-ws") .thread_name("tokio-ws")
// limit concurrent SQLite blocking threads
.max_blocking_threads(settings.limits.max_blocking_threads)
.on_thread_start(|| {
debug!("started new thread");
})
.on_thread_stop(|| {
debug!("stopping thread");
})
.build() .build()
.unwrap(); .unwrap();
// start tokio // start tokio
@@ -320,7 +311,6 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
controlled_shutdown.send(()).ok(); controlled_shutdown.send(()).ok();
} }
Err(std::sync::mpsc::RecvError) => { Err(std::sync::mpsc::RecvError) => {
// FIXME: spurious error on startup?
debug!("shutdown requestor is disconnected"); debug!("shutdown requestor is disconnected");
} }
}; };
@@ -339,7 +329,8 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
let pool = db::build_pool( let pool = db::build_pool(
"client query", "client query",
&settings, &settings,
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY
| rusqlite::OpenFlags::SQLITE_OPEN_SHARED_CACHE,
db_min_conn, db_min_conn,
db_max_conn, db_max_conn,
true, true,
@@ -418,7 +409,11 @@ fn convert_to_msg(msg: String, max_bytes: Option<usize>) -> Result<NostrMessage>
fn make_notice_message(notice: Notice) -> Message { fn make_notice_message(notice: Notice) -> Message {
let json = match notice { let json = match notice {
Notice::Message(ref msg) => json!(["NOTICE", msg]), Notice::Message(ref msg) => json!(["NOTICE", msg]),
Notice::EventResult(ref res) => json!(["OK", res.id, res.status.to_bool(), res.msg]), Notice::EventResult(ref res) => match &res.status {
EventResultStatus::Saved => json!(["OK", res.id, "true"]),
EventResultStatus::Duplicate => json!(["OK", res.id, "true", "duplicate"]),
EventResultStatus::Error(msg) => json!(["OK", res.id, "false", msg]),
},
}; };
Message::text(json.to_string()) Message::text(json.to_string())
@@ -440,34 +435,17 @@ async fn nostr_server(
event_tx: mpsc::Sender<SubmittedEvent>, event_tx: mpsc::Sender<SubmittedEvent>,
mut shutdown: Receiver<()>, mut shutdown: Receiver<()>,
) { ) {
// the time this websocket nostr server started
let orig_start = Instant::now();
// get a broadcast channel for clients to communicate on // get a broadcast channel for clients to communicate on
let mut bcast_rx = broadcast.subscribe(); let mut bcast_rx = broadcast.subscribe();
// Track internal client state // Track internal client state
let mut conn = conn::ClientConn::new(client_info.remote_ip); let mut conn = conn::ClientConn::new(client_info.remote_ip);
// subscription creation rate limiting
let mut sub_lim_opt = None;
// 100ms jitter when the rate limiter returns
let jitter = Jitter::up_to(Duration::from_millis(100));
let sub_per_min_setting = settings.limits.subscriptions_per_min;
if let Some(sub_per_min) = sub_per_min_setting {
if sub_per_min > 0 {
trace!("Rate limits for sub creation ({}/min)", sub_per_min);
let quota_time = core::num::NonZeroU32::new(sub_per_min).unwrap();
let quota = Quota::per_minute(quota_time);
sub_lim_opt = Some(RateLimiter::direct(quota));
}
}
// Use the remote IP as the client identifier // Use the remote IP as the client identifier
let cid = conn.get_client_prefix(); let cid = conn.get_client_prefix();
// Create a channel for receiving query results from the database. // Create a channel for receiving query results from the database.
// we will send out the tx handle to any query we generate. // we will send out the tx handle to any query we generate.
// this has capacity for some of the larger requests we see, which let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(256);
// should allow the DB thread to release the handle earlier.
let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(20000);
// Create channel for receiving NOTICEs // Create channel for receiving NOTICEs
let (notice_tx, mut notice_rx) = mpsc::channel::<Notice>(128); let (notice_tx, mut notice_rx) = mpsc::channel::<Notice>(32);
// last time this client sent data (message, ping, etc.) // last time this client sent data (message, ping, etc.)
let mut last_message_time = Instant::now(); let mut last_message_time = Instant::now();
@@ -485,18 +463,19 @@ async fn nostr_server(
// when these subscriptions are cancelled, make a message // when these subscriptions are cancelled, make a message
// available to the executing query so it knows to stop. // available to the executing query so it knows to stop.
let mut running_queries: HashMap<String, oneshot::Sender<()>> = HashMap::new(); let mut running_queries: HashMap<String, oneshot::Sender<()>> = HashMap::new();
// for stats, keep track of how many events the client published, // for stats, keep track of how many events the client published,
// and how many it received from queries. // and how many it received from queries.
let mut client_published_event_count: usize = 0; let mut client_published_event_count: usize = 0;
let mut client_received_event_count: usize = 0; let mut client_received_event_count: usize = 0;
debug!("new client connection (cid: {}, ip: {:?})", cid, conn.ip()); debug!("new connection for client: {}, ip: {:?}", cid, conn.ip());
if let Some(ua) = client_info.user_agent { if let Some(ua) = client_info.user_agent {
debug!("cid: {}, user-agent: {:?}", cid, ua); debug!("client: {} has user-agent: {:?}", cid, ua);
} }
loop { loop {
tokio::select! { tokio::select! {
_ = shutdown.recv() => { _ = shutdown.recv() => {
info!("Close connection down due to shutdown, client: {}, ip: {:?}, connected: {:?}", cid, conn.ip(), orig_start.elapsed()); info!("Close connection down due to shutdown, client: {}, ip: {:?}", cid, conn.ip());
// server shutting down, exit loop // server shutting down, exit loop
break; break;
}, },
@@ -577,17 +556,17 @@ async fn nostr_server(
Err(WsError::AlreadyClosed | WsError::ConnectionClosed | Err(WsError::AlreadyClosed | WsError::ConnectionClosed |
WsError::Protocol(tungstenite::error::ProtocolError::ResetWithoutClosingHandshake))) WsError::Protocol(tungstenite::error::ProtocolError::ResetWithoutClosingHandshake)))
=> { => {
debug!("websocket close from client (cid: {}, ip: {:?})",cid, conn.ip()); debug!("websocket close from client: {}, ip: {:?}",cid, conn.ip());
break; break;
}, },
Some(Err(WsError::Io(e))) => { Some(Err(WsError::Io(e))) => {
// IO errors are considered fatal // IO errors are considered fatal
warn!("IO error (cid: {}, ip: {:?}): {:?}", cid, conn.ip(), e); warn!("IO error (client: {}, ip: {:?}): {:?}", cid, conn.ip(), e);
break; break;
} }
x => { x => {
// default condition on error is to close the client connection // default condition on error is to close the client connection
info!("unknown error (cid: {}, ip: {:?}): {:?} (closing conn)", cid, conn.ip(), x); info!("unknown error (client: {}, ip: {:?}): {:?} (closing conn)", cid, conn.ip(), x);
break; break;
} }
}; };
@@ -597,12 +576,11 @@ async fn nostr_server(
Ok(NostrMessage::EventMsg(ec)) => { Ok(NostrMessage::EventMsg(ec)) => {
// An EventCmd needs to be validated to be converted into an Event // An EventCmd needs to be validated to be converted into an Event
// handle each type of message // handle each type of message
let evid = ec.event_id().to_owned();
let parsed : Result<Event> = Result::<Event>::from(ec); let parsed : Result<Event> = Result::<Event>::from(ec);
match parsed { match parsed {
Ok(e) => { Ok(e) => {
let id_prefix:String = e.id.chars().take(8).collect(); let id_prefix:String = e.id.chars().take(8).collect();
debug!("successfully parsed/validated event: {:?} (cid: {})", id_prefix, cid); debug!("successfully parsed/validated event: {:?} from client: {}", id_prefix, cid);
// check if the event is too far in the future. // check if the event is too far in the future.
if e.is_valid_timestamp(settings.options.reject_future_seconds) { if e.is_valid_timestamp(settings.options.reject_future_seconds) {
// Write this to the database. // Write this to the database.
@@ -612,48 +590,39 @@ async fn nostr_server(
} else { } else {
info!("client: {} sent a far future-dated event", cid); info!("client: {} sent a far future-dated event", cid);
if let Some(fut_sec) = settings.options.reject_future_seconds { if let Some(fut_sec) = settings.options.reject_future_seconds {
let msg = format!("The event created_at field is out of the acceptable range (+{}sec) for this relay.",fut_sec); let msg = format!("The event created_at field is out of the acceptable range (+{}sec) for this relay and was not stored.",fut_sec);
let notice = Notice::invalid(e.id, &msg); let notice = Notice::err_msg(msg, e.id);
ws_stream.send(make_notice_message(notice)).await.ok(); ws_stream.send(make_notice_message(notice)).await.ok();
} }
} }
}, },
Err(e) => { Err(_) => {
info!("client sent an invalid event (cid: {})", cid); info!("client: {} sent an invalid event", cid);
ws_stream.send(make_notice_message(Notice::invalid(evid, &format!("{}", e)))).await.ok(); ws_stream.send(make_notice_message(Notice::message("event was invalid".into()))).await.ok();
} }
} }
}, },
Ok(NostrMessage::SubMsg(s)) => { Ok(NostrMessage::SubMsg(s)) => {
debug!("subscription requested (cid: {}, sub: {:?})", cid, s.id); debug!("client: {} requesting a subscription", cid);
// subscription handling consists of: // subscription handling consists of:
// * check for rate limits
// * registering the subscription so future events can be matched // * registering the subscription so future events can be matched
// * making a channel to cancel to request later // * making a channel to cancel to request later
// * sending a request for a SQL query // * sending a request for a SQL query
// Do nothing if the sub already exists. let (abandon_query_tx, abandon_query_rx) = oneshot::channel::<()>();
if !conn.has_subscription(&s) { match conn.subscribe(s.clone()) {
if let Some(ref lim) = sub_lim_opt { Ok(()) => {
lim.until_ready_with_jitter(jitter).await; // when we insert, if there was a previous query running with the same name, cancel it.
} if let Some(previous_query) = running_queries.insert(s.id.to_owned(), abandon_query_tx) {
let (abandon_query_tx, abandon_query_rx) = oneshot::channel::<()>(); previous_query.send(()).ok();
match conn.subscribe(s.clone()) { }
Ok(()) => { // start a database query
// when we insert, if there was a previous query running with the same name, cancel it. db::db_query(s, cid.to_owned(), pool.clone(), query_tx.clone(), abandon_query_rx).await;
if let Some(previous_query) = running_queries.insert(s.id.to_owned(), abandon_query_tx) { },
previous_query.send(()).ok(); Err(e) => {
} info!("Subscription error: {}", e);
// start a database query. this spawns a blocking database query on a worker thread. ws_stream.send(make_notice_message(Notice::err(e, s.id))).await.ok();
db::db_query(s, cid.to_owned(), pool.clone(), query_tx.clone(), abandon_query_rx).await;
},
Err(e) => {
info!("Subscription error: {} (cid: {}, sub: {:?})", e, cid, s.id);
ws_stream.send(make_notice_message(Notice::message(format!("Subscription error: {}", e)))).await.ok();
}
} }
} else { }
info!("client sent duplicate subscription, ignoring (cid: {}, sub: {:?})", cid, s.id);
}
}, },
Ok(NostrMessage::CloseMsg(cc)) => { Ok(NostrMessage::CloseMsg(cc)) => {
// closing a request simply removes the subscription. // closing a request simply removes the subscription.
@@ -674,19 +643,19 @@ async fn nostr_server(
} }
}, },
Err(Error::ConnError) => { Err(Error::ConnError) => {
debug!("got connection close/error, disconnecting cid: {}, ip: {:?}",cid, conn.ip()); debug!("got connection close/error, disconnecting client: {}, ip: {:?}",cid, conn.ip());
break; break;
} }
Err(Error::EventMaxLengthError(s)) => { Err(Error::EventMaxLengthError(s)) => {
info!("client sent event larger ({} bytes) than max size (cid: {})", s, cid); info!("client: {} sent event larger ({} bytes) than max size", cid, s);
ws_stream.send(make_notice_message(Notice::message("event exceeded max size".into()))).await.ok(); ws_stream.send(make_notice_message(Notice::message("event exceeded max size".into()))).await.ok();
}, },
Err(Error::ProtoParseError) => { Err(Error::ProtoParseError) => {
info!("client sent event that could not be parsed (cid: {})", cid); info!("client {} sent event that could not be parsed", cid);
ws_stream.send(make_notice_message(Notice::message("could not parse command".into()))).await.ok(); ws_stream.send(make_notice_message(Notice::message("could not parse command".into()))).await.ok();
}, },
Err(e) => { Err(e) => {
info!("got non-fatal error from client (cid: {}, error: {:?}", cid, e); info!("got non-fatal error from client: {}, error: {:?}", cid, e);
}, },
} }
}, },
@@ -697,11 +666,10 @@ async fn nostr_server(
stop_tx.send(()).ok(); stop_tx.send(()).ok();
} }
info!( info!(
"stopping client connection (cid: {}, ip: {:?}, sent: {} events, recv: {} events, connected: {:?})", "stopping connection for client: {}, ip: {:?} (client sent {} event(s), received {})",
cid, cid,
conn.ip(), conn.ip(),
client_published_event_count, client_published_event_count,
client_received_event_count, client_received_event_count
orig_start.elapsed()
); );
} }
-3
View File
@@ -37,9 +37,6 @@ pub struct ReqFilter {
#[serde(skip)] #[serde(skip)]
pub tags: Option<HashMap<char, HashSet<String>>>, pub tags: Option<HashMap<char, HashSet<String>>>,
/// Force no matches due to malformed data /// Force no matches due to malformed data
// we can't represent it in the req filter, so we don't want to
// erroneously match. This basically indicates the req tried to
// do something invalid.
pub force_no_match: bool, pub force_no_match: bool,
} }