diff --git a/Cargo.lock b/Cargo.lock index f0fa5bbe..0a3fa7f7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -294,6 +294,15 @@ dependencies = [ "syn", ] +[[package]] +name = "atoi" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528" +dependencies = [ + "num-traits", +] + [[package]] name = "atomic" version = "0.6.1" @@ -490,6 +499,9 @@ name = "bitflags" version = "2.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" +dependencies = [ + "serde_core", +] [[package]] name = "block-buffer" @@ -500,6 +512,15 @@ dependencies = [ "generic-array", ] +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + [[package]] name = "block2" version = "0.6.2" @@ -552,6 +573,12 @@ version = "1.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec" +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.12.0" @@ -634,6 +661,12 @@ dependencies = [ "cc", ] +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + [[package]] name = "combine" version = "4.6.7" @@ -732,6 +765,21 @@ dependencies = [ "libc", ] +[[package]] +name = "crc" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d" +dependencies = [ + "crc-catalog", +] + +[[package]] +name = "crc-catalog" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" + [[package]] name = "crc32fast" version = "1.5.0" @@ -765,6 +813,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-queue" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "803d13fb3b09d88be9f4dbc29062c66b19bf7170867ceb746d2a8689bf6c7a26" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -793,6 +850,24 @@ dependencies = [ "typenum", ] +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "ctutils" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d5515a3834141de9eafb9717ad39eea8247b5674e6066c404e8c4b365d2a29e" +dependencies = [ + "cmov", +] + [[package]] name = "curve25519-dalek" version = "4.1.3" @@ -802,7 +877,7 @@ dependencies = [ "cfg-if", "cpufeatures 0.2.17", "curve25519-dalek-derive", - "digest", + "digest 0.10.7", "fiat-crypto", "rustc_version", "subtle", @@ -929,12 +1004,23 @@ version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ - "block-buffer", + "block-buffer 0.10.4", "const-oid", - "crypto-common", + "crypto-common 0.1.6", "subtle", ] +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer 0.12.1", + "crypto-common 0.2.2", + "ctutils", +] + [[package]] name = "dispatch2" version = "0.3.1" @@ -956,6 +1042,12 @@ dependencies = [ "syn", ] +[[package]] +name = "dotenvy" +version = "0.15.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" + [[package]] name = "dunce" version = "1.0.5" @@ -975,7 +1067,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ee27f32b5c5292967d2d4a9d7f1e0b0aed2c15daded5a60300e4abb9d8020bca" dependencies = [ "der 0.7.10", - "digest", + "digest 0.10.7", "elliptic-curve", "rfc6979", "signature", @@ -1001,7 +1093,7 @@ dependencies = [ "curve25519-dalek", "ed25519", "serde", - "sha2", + "sha2 0.10.9", "subtle", "zeroize", ] @@ -1011,6 +1103,9 @@ name = "either" version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +dependencies = [ + "serde", +] [[package]] name = "elegant-departure" @@ -1032,11 +1127,11 @@ checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47" dependencies = [ "base16ct", "crypto-bigint", - "digest", + "digest 0.10.7", "ff", "generic-array", "group", - "hkdf", + "hkdf 0.12.4", "pem-rfc7468 0.7.0", "pkcs8", "rand_core 0.6.4", @@ -1102,6 +1197,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "etcetera" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de48cc4d1c1d97a20fd819def54b890cadde72ed3ad0c614822a0a433361be96" +dependencies = [ + "cfg-if", + "windows-sys 0.61.2", +] + +[[package]] +name = "event-listener" +version = "5.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] + [[package]] name = "fancy-regex" version = "0.16.2" @@ -1196,6 +1311,17 @@ dependencies = [ "serde", ] +[[package]] +name = "flume" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1296,6 +1422,17 @@ dependencies = [ "futures-util", ] +[[package]] +name = "futures-intrusive" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f" +dependencies = [ + "futures-core", + "lock_api", + "parking_lot", +] + [[package]] name = "futures-io" version = "0.3.32" @@ -1505,6 +1642,15 @@ dependencies = [ "foldhash 0.2.0", ] +[[package]] +name = "hashlink" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824e001ac4f3012dd16a264bec811403a67ca9deb6c102fc5049b32c4574b35f" +dependencies = [ + "hashbrown 0.16.1", +] + [[package]] name = "heck" version = "0.5.0" @@ -1599,7 +1745,16 @@ version = "0.12.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7" dependencies = [ - "hmac", + "hmac 0.12.1", +] + +[[package]] +name = "hkdf" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4aaa26c720c68b866f2c96ef5c1264b3e6f473fe5d4ce61cd44bbe913e553018" +dependencies = [ + "hmac 0.13.0", ] [[package]] @@ -1608,7 +1763,16 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" dependencies = [ - "digest", + "digest 0.10.7", +] + +[[package]] +name = "hmac" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" +dependencies = [ + "digest 0.11.3", ] [[package]] @@ -1694,6 +1858,15 @@ dependencies = [ "serde", ] +[[package]] +name = "hybrid-array" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" +dependencies = [ + "typenum", +] + [[package]] name = "hyper" version = "1.10.1" @@ -2155,7 +2328,7 @@ dependencies = [ "base64", "ed25519-dalek", "getrandom 0.2.17", - "hmac", + "hmac 0.12.1", "js-sys", "p256", "p384", @@ -2164,7 +2337,7 @@ dependencies = [ "rsa", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "signature", "simple_asn1", "zeroize", @@ -2197,6 +2370,17 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "libsqlite3-sys" +version = "0.37.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f111c8c41e7c61a49cd34e44c7619462967221a6443b0ec299e0ac30cfb9b1" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -2273,6 +2457,16 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest 0.11.3", +] + [[package]] name = "mediatype" version = "0.21.0" @@ -2841,6 +3035,7 @@ dependencies = [ "sentry", "serde", "serde_json", + "sqlx", "stresstest", "tempfile", "thiserror", @@ -2876,6 +3071,7 @@ dependencies = [ "sentry", "serde", "serde_json", + "sqlx", "tempfile", "thiserror", "tokio", @@ -3028,7 +3224,7 @@ dependencies = [ "ecdsa", "elliptic-curve", "primeorder", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -3040,7 +3236,7 @@ dependencies = [ "ecdsa", "elliptic-curve", "primeorder", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -3053,6 +3249,12 @@ dependencies = [ "seize", ] +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -3817,7 +4019,7 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2" dependencies = [ - "hmac", + "hmac 0.12.1", "subtle", ] @@ -3842,7 +4044,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d" dependencies = [ "const-oid", - "digest", + "digest 0.10.7", "num-bigint-dig", "num-integer", "num-traits", @@ -4194,7 +4396,7 @@ dependencies = [ "num", "sentry-options-validation", "serde_json", - "sha1", + "sha1 0.10.6", "thiserror", ] @@ -4388,7 +4590,18 @@ checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" dependencies = [ "cfg-if", "cpufeatures 0.2.17", - "digest", + "digest 0.10.7", +] + +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.3", ] [[package]] @@ -4399,7 +4612,18 @@ checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", "cpufeatures 0.2.17", - "digest", + "digest 0.10.7", +] + +[[package]] +name = "sha2" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.3", ] [[package]] @@ -4433,7 +4657,7 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ - "digest", + "digest 0.10.7", "rand_core 0.6.4", ] @@ -4494,6 +4718,9 @@ name = "smallvec" version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +dependencies = [ + "serde", +] [[package]] name = "socket2" @@ -4520,6 +4747,9 @@ name = "spin" version = "0.9.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] [[package]] name = "spki" @@ -4531,6 +4761,183 @@ dependencies = [ "der 0.7.10", ] +[[package]] +name = "sqlx" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "378620ccc25c62c89d8be1c819e76a88d59bdcc3304733330788948e619bfd71" +dependencies = [ + "sqlx-core", + "sqlx-macros", + "sqlx-mysql", + "sqlx-postgres", + "sqlx-sqlite", +] + +[[package]] +name = "sqlx-core" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb" +dependencies = [ + "base64", + "bytes", + "cfg-if", + "chrono", + "crc", + "crossbeam-queue", + "either", + "event-listener", + "futures-core", + "futures-intrusive", + "futures-io", + "futures-util", + "hashbrown 0.16.1", + "hashlink", + "indexmap 2.14.0", + "log", + "memchr", + "percent-encoding", + "rustls", + "rustls-native-certs", + "serde", + "serde_json", + "sha2 0.10.9", + "smallvec", + "thiserror", + "tokio", + "tokio-stream", + "tracing", + "url", +] + +[[package]] +name = "sqlx-macros" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bd2b84f2bc39a5705ef27ec785a11c934a41bbd4a24941e257927cddc26b60bf" +dependencies = [ + "proc-macro2", + "quote", + "sqlx-core", + "sqlx-macros-core", + "syn", +] + +[[package]] +name = "sqlx-macros-core" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb8d96de5fdc85a5c4ec813432b523ec637e80ba98f046555f75f7908ddac7c3" +dependencies = [ + "cfg-if", + "dotenvy", + "either", + "heck", + "hex", + "proc-macro2", + "quote", + "serde", + "serde_json", + "sha2 0.10.9", + "sqlx-core", + "sqlx-mysql", + "sqlx-postgres", + "sqlx-sqlite", + "syn", + "thiserror", + "tokio", + "url", +] + +[[package]] +name = "sqlx-mysql" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90b8020fe17c5f2c245bfa2505d7ef59c5604839527c740266ad2214acebea27" +dependencies = [ + "bitflags", + "byteorder", + "bytes", + "chrono", + "crc", + "digest 0.11.3", + "dotenvy", + "either", + "futures-core", + "futures-util", + "generic-array", + "log", + "percent-encoding", + "serde", + "sha1 0.11.0", + "sha2 0.11.0", + "sqlx-core", + "thiserror", + "tracing", +] + +[[package]] +name = "sqlx-postgres" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e" +dependencies = [ + "atoi", + "base64", + "bitflags", + "byteorder", + "chrono", + "crc", + "dotenvy", + "etcetera", + "futures-channel", + "futures-core", + "futures-util", + "hex", + "hkdf 0.13.0", + "hmac 0.13.0", + "itoa", + "log", + "md-5", + "memchr", + "rand 0.10.2", + "serde", + "serde_json", + "sha2 0.11.0", + "smallvec", + "sqlx-core", + "stringprep", + "thiserror", + "tracing", + "whoami", +] + +[[package]] +name = "sqlx-sqlite" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488e99c397a62007e4229aec669a179816339afc6d2620ca6fa420dbee2e982c" +dependencies = [ + "atoi", + "chrono", + "flume", + "form_urlencoded", + "futures-channel", + "futures-core", + "futures-executor", + "futures-intrusive", + "futures-util", + "libsqlite3-sys", + "log", + "percent-encoding", + "serde", + "sqlx-core", + "thiserror", + "tracing", + "url", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -4561,6 +4968,17 @@ dependencies = [ "yansi", ] +[[package]] +name = "stringprep" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1" +dependencies = [ + "unicode-bidi", + "unicode-normalization", + "unicode-properties", +] + [[package]] name = "strsim" version = "0.11.1" @@ -5121,6 +5539,12 @@ version = "2.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" +[[package]] +name = "unicode-bidi" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5" + [[package]] name = "unicode-general-category" version = "1.1.0" @@ -5133,6 +5557,21 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-normalization" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fd4f6878c9cb28d874b009da9e8d183b5abc80117c40bbd187a1fde336be6e8" +dependencies = [ + "tinyvec", +] + +[[package]] +name = "unicode-properties" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" + [[package]] name = "unicode-segmentation" version = "1.13.3" @@ -5400,6 +5839,12 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "whoami" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "998767ef88740d1f5b0682a9c53c24431453923962269c2db68ee43788c5a40d" + [[package]] name = "widestring" version = "1.2.1" diff --git a/Cargo.toml b/Cargo.toml index 83c9d77c..fbe423ca 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -83,12 +83,21 @@ sentry-options = "1.2.1" serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.150" serde_yaml = "0.9.34-deprecated" +sqlx = { version = "0.9.0", features = [ + "runtime-tokio", + "tls-rustls-ring-native-roots", + "sqlite", + "chrono", +] } sketches-ddsketch = "0.3.1" tempfile = "3.27.0" thiserror = "2.0.18" thread_local = "1.1.9" jemalloc_pprof = "0.9.0" -tikv-jemallocator = { version = "0.7.0", features = ["background_threads", "override_allocator_on_supported_platforms"] } +tikv-jemallocator = { version = "0.7.0", features = [ + "background_threads", + "override_allocator_on_supported_platforms", +] } tikv-jemalloc-ctl = { version = "0.7.0", features = ["stats"] } tokio = "1.52.3" tokio-util = "0.7.18" diff --git a/migrations/sqlite/0001_keeper.sql b/migrations/sqlite/0001_keeper.sql new file mode 100644 index 00000000..b3add96a --- /dev/null +++ b/migrations/sqlite/0001_keeper.sql @@ -0,0 +1,7 @@ +CREATE TABLE IF NOT EXISTS ttl_keeper ( + object_id TEXT PRIMARY KEY, + expiration_policy INTEGER NOT NULL DEFAULT 0, + duration INTEGER, + created_at INTEGER NOT NULL, + expires_at INTEGER +); diff --git a/objectstore-server/Cargo.toml b/objectstore-server/Cargo.toml index d6ddec62..18cdbafd 100644 --- a/objectstore-server/Cargo.toml +++ b/objectstore-server/Cargo.toml @@ -38,17 +38,40 @@ papaya = { workspace = true } pin-project-lite = { workspace = true } percent-encoding = { workspace = true } rand = { workspace = true } -reqwest = { workspace = true, features = ["charset", "http2", "system-proxy", "native-tls-no-alpn"] } +reqwest = { workspace = true, features = [ + "charset", + "http2", + "system-proxy", + "native-tls-no-alpn", +] } rustls = { workspace = true } secrecy = { workspace = true, features = ["serde"] } -sentry = { workspace = true, features = ["tower-axum-matched-path", "tracing", "logs"] } +sentry = { workspace = true, features = [ + "tower-axum-matched-path", + "tracing", + "logs", +] } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } thread_local = { workspace = true } -tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } +tokio = { workspace = true, features = [ + "fs", + "macros", + "net", + "rt-multi-thread", + "signal", + "sync", + "time", +] } tower = { workspace = true } -tower-http = { workspace = true, features = ["catch-panic", "metrics", "set-header", "trace"] } +tower-http = { workspace = true, features = [ + "catch-panic", + "metrics", + "set-header", + "trace", +] } +sqlx.workspace = true [target.'cfg(target_os = "linux")'.dependencies] jemalloc_pprof = { workspace = true, optional = true } diff --git a/objectstore-service/Cargo.toml b/objectstore-service/Cargo.toml index df6b771a..3564f38c 100644 --- a/objectstore-service/Cargo.toml +++ b/objectstore-service/Cargo.toml @@ -25,10 +25,20 @@ objectstore-metrics = { workspace = true } objectstore-types = { workspace = true } quick-xml = { workspace = true, features = ["serialize"] } regex = { workspace = true } -reqwest = { workspace = true, features = ["charset", "http2", "hickory-dns", "json", "multipart", "native-tls-no-alpn", "stream", "system-proxy"] } +reqwest = { workspace = true, features = [ + "charset", + "http2", + "hickory-dns", + "json", + "multipart", + "native-tls-no-alpn", + "stream", + "system-proxy", +] } sentry = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +sqlx = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } tokio-util = { workspace = true, features = ["io", "rt"] } diff --git a/objectstore-service/src/error.rs b/objectstore-service/src/error.rs index 9643dd43..21505a1e 100644 --- a/objectstore-service/src/error.rs +++ b/objectstore-service/src/error.rs @@ -151,6 +151,14 @@ pub enum Error { /// Invalid upload ID (e.g. path traversal attempt). #[error(transparent)] InvalidUploadId(#[from] objectstore_types::multipart::InvalidUploadId), + + /// Errors from SQLx database migrations. + #[error("sqlx migration error: {0}")] + SqlxMigrate(#[from] sqlx::migrate::MigrateError), + + /// Common SQLx database errors. + #[error("sqlx error: {0}")] + Sqlx(#[from] sqlx::Error), } impl Error { @@ -211,6 +219,8 @@ impl Error { Self::NotImplemented => Level::ERROR, Self::InvalidUploadId(_) => Level::DEBUG, Self::Generic { .. } => Level::ERROR, + Self::SqlxMigrate(_) => Level::ERROR, + Self::Sqlx(_) => Level::ERROR, } } } diff --git a/objectstore-service/src/keeper/mod.rs b/objectstore-service/src/keeper/mod.rs new file mode 100644 index 00000000..85661629 --- /dev/null +++ b/objectstore-service/src/keeper/mod.rs @@ -0,0 +1,34 @@ +//! Keeper defines a trait to support extending an object retention for backends +//! that does not support them natively (e.g., S3-compatible backend and filesystem). + +use objectstore_types::metadata::ExpirationPolicy; + +use crate::error::Result; +use crate::id::ObjectId; + +/// SQLite-backed implementation of the [`Keeper`] trait. +pub mod sqlite_backed; + +/// Represents the computed expiry information for an object. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObjectExpiry<'a> { + id: &'a ObjectId, + expiration_policy: ExpirationPolicy, +} + +/// Object retention keeper trait. +#[async_trait::async_trait] +pub trait Keeper: Send + Sync { + /// Keep is the first step in the object retention lifecycle. + /// Practically speaking, it would not be kept if the `expiration_policy` is `Manual`. + /// The `expiration_policy` is set by the client at upload time via the supplied Metadata. + async fn keep(&self, id: &ObjectId, expiration_policy: ExpirationPolicy) -> Result<()>; + + /// Remove is the final step in the object retention lifecycle. + /// It is called by a cleanup worker when the object is no longer needed. + async fn remove(&self, id: &ObjectId) -> Result<()>; + + /// Marks an object as accessed. For `expiration_policy` of `TimeToIdle`, this will + /// extend the object retention. + async fn mark_accessed(&self, id: &ObjectId) -> Result<()>; +} diff --git a/objectstore-service/src/keeper/sqlite_backed.rs b/objectstore-service/src/keeper/sqlite_backed.rs new file mode 100644 index 00000000..f8cd9ddf --- /dev/null +++ b/objectstore-service/src/keeper/sqlite_backed.rs @@ -0,0 +1,456 @@ +use std::str::FromStr; +use std::time::{SystemTime, UNIX_EPOCH}; + +use sqlx::migrate::MigrateDatabase; +use sqlx::pool::PoolOptions; +use sqlx::sqlite::{ + SqliteAutoVacuum, SqliteConnectOptions, SqliteJournalMode, SqlitePool, SqliteSynchronous, +}; +use sqlx::{ConnectOptions, FromRow, Pool, Sqlite}; + +use objectstore_types::metadata::ExpirationPolicy; + +use crate::error::{Error, Result}; +use crate::id::ObjectId; + +/// A row from the `ttl_keeper` table. +#[derive(Debug, FromRow)] +pub struct TableRow { + /// The storage path key for the object. + pub object_id: String, + /// Encoded expiration policy: 0 = Manual, 1 = TimeToLive, 2 = TimeToIdle. + pub expiration_policy: i32, + /// The expiration duration in seconds, if applicable. + pub duration: Option, + /// Unix timestamp (seconds) when the row was created. + pub created_at: i64, + /// Unix timestamp (seconds) when the object expires, if applicable. + pub expires_at: Option, +} + +/// SQLite-backed keeper that persists object retention state in a `ttl_keeper` table. +#[derive(Debug)] +pub struct SqliteBackedKeeper { + read_pool: SqlitePool, + write_pool: SqlitePool, +} + +/// Creates a pair of read/write SQLite connection pools for the given URL. +pub async fn create_sqlite_pool(url: &str) -> Result<(Pool, Pool)> { + if !Sqlite::database_exists(url).await? { + Sqlite::create_database(url).await? + } + + let read_pool = PoolOptions::::new() + .max_connections(64) + .connect_with( + SqliteConnectOptions::from_str(url)? + .journal_mode(SqliteJournalMode::Wal) + .synchronous(SqliteSynchronous::Normal) + .read_only(true) + .disable_statement_logging(), + ) + .await?; + + let write_pool = PoolOptions::::new() + .max_connections(1) + .connect_with( + SqliteConnectOptions::from_str(url)? + .journal_mode(SqliteJournalMode::Wal) + .synchronous(SqliteSynchronous::Normal) + .auto_vacuum(SqliteAutoVacuum::Incremental) + .disable_statement_logging(), + ) + .await?; + + Ok((read_pool, write_pool)) +} + +impl SqliteBackedKeeper { + /// Creates a new [`SqliteBackedKeeper`], running migrations against the given URL. + pub async fn new(connection_url: &str) -> Result { + let (read_pool, write_pool) = create_sqlite_pool(connection_url).await?; + + sqlx::migrate!("./../migrations/sqlite") + .run(&write_pool) + .await?; + + Ok(Self { + read_pool, + write_pool, + }) + } +} + +#[async_trait::async_trait] +impl super::Keeper for SqliteBackedKeeper { + async fn keep(&self, id: &ObjectId, expiration_policy: ExpirationPolicy) -> Result<()> { + if expiration_policy.is_manual() { + return Ok(()); + } + + let expiration_duration: Option = expiration_policy.expires_in().and_then(|x| { + x.as_secs() + .try_into() + .map_err(|_| Error::generic("expiration duration exceeds i64::MAX")) + .ok() + }); + + let current_time: i64 = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| Error::generic("system time before UNIX_EPOCH"))? + .as_secs() + .try_into() + .map_err(|_| Error::generic("current time exceeds i64::MAX"))?; + + let expires_at = expiration_duration.map(|duration| current_time + duration); + + let mut atomic = self.write_pool.begin().await?; + sqlx::query( + " + INSERT INTO ttl_keeper (object_id, expiration_policy, duration, created_at, expires_at) + VALUES (?, ?, ?, ?, ?) + ", + ) + .bind(id.as_storage_path().to_string()) + .bind(match expiration_policy { + ExpirationPolicy::Manual => 0, + ExpirationPolicy::TimeToLive(_) => 1, + ExpirationPolicy::TimeToIdle(_) => 2, + }) + .bind(expiration_duration) + .bind(current_time) + .bind(expires_at) + .execute(&mut *atomic) + .await?; + + atomic.commit().await?; + + Ok(()) + } + + async fn remove(&self, id: &ObjectId) -> Result<()> { + let mut atomic = self.write_pool.begin().await?; + sqlx::query( + " + DELETE FROM ttl_keeper + WHERE object_id = ? + ", + ) + .bind(id.as_storage_path().to_string()) + .execute(&mut *atomic) + .await?; + + atomic.commit().await?; + + Ok(()) + } + + async fn mark_accessed(&self, id: &ObjectId) -> Result<()> { + // Check what expiration policy is set for this object. + let expiration_policy: Option = sqlx::query_as( + " + SELECT + object_id, + expiration_policy, + duration, + created_at, + expires_at + FROM ttl_keeper + WHERE object_id = ? + ", + ) + .bind(id.as_storage_path().to_string()) + .fetch_optional(&self.read_pool) + .await?; + + if let Some(expiration_policy) = expiration_policy { + // We only check if the object's policy is TimeToIdle. + // If it is, we need to extend the object's time to live. + // For other policies, we do nothing. + if expiration_policy.expiration_policy == 2 { + // `expiration_policy.duration` should not be `None`. + // If it is, we should not be here. + let duration = match expiration_policy.duration { + Some(duration) => duration, + None => { + return Err(Error::generic( + "unwanted state for time to idle: duration is None", + )); + } + }; + + // If `expiration_policy.expires_at` is None, we should set it to `current_time + duration`. + let current_time: i64 = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| Error::generic("system time before UNIX_EPOCH"))? + .as_secs() + .try_into() + .map_err(|_| Error::generic("current time exceeds i64::MAX"))?; + + // Update the object's expiration time. + let mut atomic = self.write_pool.begin().await?; + sqlx::query( + " + UPDATE ttl_keeper + SET expires_at = ? + WHERE object_id = ? + ", + ) + .bind(current_time + duration) + .bind(id.as_storage_path().to_string()) + .execute(&mut *atomic) + .await?; + + atomic.commit().await?; + } + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use sqlx::sqlite::SqlitePool; + + use objectstore_types::metadata::ExpirationPolicy; + use objectstore_types::scope::{Scope, Scopes}; + + use crate::id::{ObjectContext, ObjectId}; + use crate::keeper::Keeper; + + use super::*; + + fn make_id() -> ObjectId { + let context = ObjectContext { + usecase: "testing".into(), + scopes: Scopes::from_iter([Scope::create("testing", "value").unwrap()]), + }; + ObjectId::random(context) + } + + struct TestKeeper { + keeper: SqliteBackedKeeper, + pool: SqlitePool, + } + + impl TestKeeper { + async fn new() -> Self { + let pool = PoolOptions::::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .expect("failed to create in-memory pool"); + + sqlx::migrate!("./../migrations/sqlite") + .run(&pool) + .await + .expect("failed to run migrations"); + + let keeper = SqliteBackedKeeper { + read_pool: pool.clone(), + write_pool: pool.clone(), + }; + + Self { keeper, pool } + } + + async fn fetch_row(&self, id: &ObjectId) -> Option { + sqlx::query_as( + " + SELECT object_id, expiration_policy, duration, created_at, expires_at + FROM ttl_keeper + WHERE object_id = ? + ", + ) + .bind(id.as_storage_path().to_string()) + .fetch_optional(&self.pool) + .await + .expect("failed to fetch row") + } + } + + #[tokio::test] + async fn keep_manual_is_noop() { + let tk = TestKeeper::new().await; + let id = make_id(); + + tk.keeper.keep(&id, ExpirationPolicy::Manual).await.unwrap(); + assert!(tk.fetch_row(&id).await.is_none()); + } + + #[tokio::test] + async fn keep_ttl_inserts_row() { + let tk = TestKeeper::new().await; + let id = make_id(); + + let before = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() as i64; + + tk.keeper + .keep(&id, ExpirationPolicy::TimeToLive(Duration::from_secs(60))) + .await + .unwrap(); + + let row = tk.fetch_row(&id).await.expect("row should exist"); + assert_eq!(row.expiration_policy, 1); + assert_eq!(row.duration, Some(60)); + assert_eq!(row.expires_at, Some(row.created_at + 60)); + assert!(row.created_at >= before && row.created_at <= before + 2); + } + + #[tokio::test] + async fn keep_tti_inserts_row() { + let tk = TestKeeper::new().await; + let id = make_id(); + + tk.keeper + .keep(&id, ExpirationPolicy::TimeToIdle(Duration::from_secs(60))) + .await + .unwrap(); + + let row = tk.fetch_row(&id).await.expect("row should exist"); + assert_eq!(row.expiration_policy, 2); + assert_eq!(row.duration, Some(60)); + assert_eq!(row.expires_at, Some(row.created_at + 60)); + } + + #[tokio::test] + async fn remove_deletes_existing_row() { + let tk = TestKeeper::new().await; + let id = make_id(); + + tk.keeper + .keep(&id, ExpirationPolicy::TimeToLive(Duration::from_secs(60))) + .await + .unwrap(); + assert!(tk.fetch_row(&id).await.is_some()); + + tk.keeper.remove(&id).await.unwrap(); + assert!(tk.fetch_row(&id).await.is_none()); + } + + #[tokio::test] + async fn remove_nonexistent_is_ok() { + let tk = TestKeeper::new().await; + let id = make_id(); + + tk.keeper.remove(&id).await.unwrap(); + } + + #[tokio::test] + async fn mark_accessed_manual_is_noop() { + let tk = TestKeeper::new().await; + let id = make_id(); + + tk.keeper.keep(&id, ExpirationPolicy::Manual).await.unwrap(); + tk.keeper.mark_accessed(&id).await.unwrap(); + assert!(tk.fetch_row(&id).await.is_none()); + } + + #[tokio::test] + async fn mark_accessed_ttl_is_noop() { + let tk = TestKeeper::new().await; + let id = make_id(); + + tk.keeper + .keep(&id, ExpirationPolicy::TimeToLive(Duration::from_secs(60))) + .await + .unwrap(); + let before = tk.fetch_row(&id).await.unwrap(); + + tk.keeper.mark_accessed(&id).await.unwrap(); + let after = tk.fetch_row(&id).await.unwrap(); + + assert_eq!(before.expires_at, after.expires_at); + assert_eq!(before.created_at, after.created_at); + assert_eq!(before.duration, after.duration); + } + + #[tokio::test] + async fn mark_accessed_tti_without_expires_at_sets_it() { + let tk = TestKeeper::new().await; + let id = make_id(); + let id_str = id.as_storage_path().to_string(); + + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() as i64; + + // Insert a TTI row directly with expires_at = NULL. + sqlx::query( + " + INSERT INTO ttl_keeper (object_id, expiration_policy, duration, created_at, expires_at) + VALUES (?, 2, 60, ?, NULL) + ", + ) + .bind(&id_str) + .bind(now) + .execute(&tk.pool) + .await + .unwrap(); + + tk.keeper.mark_accessed(&id).await.unwrap(); + + let row = tk.fetch_row(&id).await.unwrap(); + assert_eq!(row.expires_at, Some(now + 60)); + } + + #[tokio::test] + async fn mark_accessed_tti_with_expires_at_bumps_it() { + let tk = TestKeeper::new().await; + let id = make_id(); + + tk.keeper + .keep(&id, ExpirationPolicy::TimeToIdle(Duration::from_secs(60))) + .await + .unwrap(); + let original = tk.fetch_row(&id).await.unwrap(); + + // Small delay so the timestamp changes. + tokio::time::sleep(Duration::from_millis(50)).await; + + tk.keeper.mark_accessed(&id).await.unwrap(); + let updated = tk.fetch_row(&id).await.unwrap(); + + // expires_at is always recomputed from current time + duration. + assert!(updated.expires_at.unwrap() >= original.expires_at.unwrap()); + assert_eq!(updated.created_at, original.created_at); + } + + #[tokio::test] + async fn mark_accessed_nonexistent_is_ok() { + let tk = TestKeeper::new().await; + let id = make_id(); + + tk.keeper.mark_accessed(&id).await.unwrap(); + } + + #[tokio::test] + async fn keep_duplicate_returns_error() { + let tk = TestKeeper::new().await; + let id = make_id(); + + tk.keeper + .keep(&id, ExpirationPolicy::TimeToLive(Duration::from_secs(60))) + .await + .unwrap(); + + let err = tk + .keeper + .keep(&id, ExpirationPolicy::TimeToLive(Duration::from_secs(60))) + .await + .unwrap_err(); + + assert!( + matches!(err, Error::Sqlx(_)), + "expected Error::Sqlx, got: {err:?}" + ); + } +} diff --git a/objectstore-service/src/lib.rs b/objectstore-service/src/lib.rs index f2e2eb5c..406961db 100644 --- a/objectstore-service/src/lib.rs +++ b/objectstore-service/src/lib.rs @@ -7,6 +7,7 @@ mod concurrency; pub mod error; mod gcp_auth; pub mod id; +pub mod keeper; pub mod multipart; pub mod service; pub mod stream;