comparing 64d240c819bede7f3adbcb72ced249fceef2637f and migrate on ptr.pet/nsid-tracker

+3 -12

README.md

···

       1
       1
       -
       a webapp and server that monitors the jetstream and tracks the different

     

       2
       2
       -
       lexicons as they are created or deleted. it shows you which collections are most

     

       3
       3
       -
       active on the network.

     

       1
       1
       +
       a webapp and server that monitors the jetstream and tracks the different lexicons as they are created or deleted.

     

       2
       2
       +
       it shows you which collections are most active on the network.

     

       4
       3
        
       

     

       5
       4
        
       for backend it uses rust with fjall as db, the frontend is built with sveltekit.

     

       6
       5
        
       

     

       7
       6
        
       see [here](https://gaze.systems/nsid-tracker) for a hosted instance of it.

     

       8
       7
        
       

     

       9
       9
       -
       ## performance / storage

     

       10
       10
       -
       

     

       11
       11
       -
       it uses about 50MB of space for 620M recorded events (events being just

     

       12
       12
       -
       timestamp in seconds and deleted boolean for now). and around 50-60ms for

     

       13
       13
       -
       querying 300-400k events.

     

       14
       14
       -
       

     

       15
       15
       -
       this is on a machine with AMD EPYC 7281 (32) @ 2.100GHz.

     

       16
       16
       -
       

     

       17
       8
        
       ## running

     

       18
       9
        
       

     

       19
       10
        
       ### with nix

     

       20
       11
        
       

     

       21
       21
       -
       - build the server: `nix build git+https://tangled.sh/@poor.dog/nsid-tracker#server`

     

       12
       12
       +
       - run the server: `nix run git+https://tangled.sh/@poor.dog/nsid-tracker#server`

     

       22
       13
        
       - build the client: `nix build git+https://tangled.sh/@poor.dog/nsid-tracker#client`

     

       23
       14
        
       

     

       24
       15
        
       ### manually

+2 -1

client/src/lib/components/StatsCard.svelte

···

       1
       1
        
       <script lang="ts">

     

       2
       2
        
           import { formatNumber } from "$lib/format";

     

       3
       3
       +
           import NumberFlow from "@number-flow/svelte";

     

       3
       4
        
       

     

       4
       5
        
           const colorClasses = {

     

       5
       6
        
               green: {

     
···

       44
       45
        
               {title}

     

       45
       46
        
           </h3>

     

       46
       47
        
           <p class="text-xl md:text-2xl font-bold {colors.valueText}">

     

       47
       47
       -
               {formatNumber(value)}

     

       48
       48
       +
               <NumberFlow {value} />

     

       48
       49
        
           </p>

     

       49
       50
        
       </div>

+1 -5

client/src/lib/format.ts

···

       2
       2
        
         return num.toLocaleString();

     

       3
       3
        
       };

     

       4
       4
        
       

     

       5
       5
       -
       const isValidDate = (d: Date) => d instanceof Date && !isNaN(d.getTime());

     

       6
       5
        
       export const formatTimestamp = (timestamp: number): string => {

     

       7
       7
       -
         const date = new Date(timestamp * 1000);

     

       8
       8
       -
         return isValidDate(date)

     

       9
       9
       -
           ? date.toLocaleString()

     

       10
       10
       -
           : new Date(timestamp / 1000).toLocaleString();

     

       6
       6
       +
         return new Date(timestamp / 1000).toLocaleString();

     

       11
       7
        
       };

+2 -3

client/src/routes/+layout.ts

···

       1
       1
       -
       export const prerender = false;

     

       2
       2
       -
       export const ssr = true;

     

       3
       3
       -
       export const csr = true;

     

       1
       1
       +
       export const prerender = true;

     

       2
       2
       +
       export const ssr = false;

-7

client/src/routes/+page.server.ts

···

       1
       1
       -
       import { fetchEvents, fetchTrackingSince } from "$lib/api";

     

       2
       2
       -
       

     

       3
       3
       -
       export const load = async () => {

     

       4
       4
       -
         const events = await fetchEvents();

     

       5
       5
       -
         const trackingSince = await fetchTrackingSince();

     

       6
       6
       -
         return { events, trackingSince };

     

       7
       7
       -
       };

+25 -30

client/src/routes/+page.svelte

···

       1
       1
        
       <script lang="ts">

     

       2
       2
        
           import { dev } from "$app/environment";

     

       3
       3
       -
           import type {

     

       4
       4
       -
               EventRecord,

     

       5
       5
       -
               Events,

     

       6
       6
       -
               NsidCount,

     

       7
       7
       -
               Since,

     

       8
       8
       -
               SortOption,

     

       9
       9
       -
           } from "$lib/types";

     

       3
       3
       +
           import type { EventRecord, NsidCount, SortOption } from "$lib/types";

     

       10
       4
        
           import { onMount, onDestroy } from "svelte";

     

       11
       5
        
           import { writable } from "svelte/store";

     

       12
       6
        
           import { PUBLIC_API_URL } from "$env/static/public";

     
···

       21
       15
        
           import RefreshControl from "$lib/components/RefreshControl.svelte";

     

       22
       16
        
           import { formatTimestamp } from "$lib/format";

     

       23
       17
        
       

     

       24
       24
       -
           type Props = {

     

       25
       25
       -
               data: { events: Events; trackingSince: Since };

     

       26
       26
       -
           };

     

       27
       27
       -
       

     

       28
       28
       -
           const { data }: Props = $props();

     

       29
       29
       -
       

     

       30
       30
       -
           const events = writable(

     

       31
       31
       -
               new Map<string, EventRecord>(Object.entries(data.events.events)),

     

       32
       32
       -
           );

     

       18
       18
       +
           const events = writable(new Map<string, EventRecord>());

     

       33
       19
        
           const pendingUpdates = new Map<string, EventRecord>();

     

       34
       20
        
           let eventsList: NsidCount[] = $state([]);

     

       35
       21
        
           let updateTimer: NodeJS.Timeout | null = null;

     
···

       42
       28
        
                   }))

     

       43
       29
        
                   .toArray();

     

       44
       30
        
           });

     

       45
       45
       -
           let per_second = $state(data.events.per_second);

     

       46
       46
       -
           let tracking_since = $state(data.trackingSince.since);

     

       47
       47
       -
       

     

       48
       48
       -
           const applyEvents = (newEvents: Record<string, EventRecord>) => {

     

       49
       49
       -
               events.update((map) => {

     

       50
       50
       -
                   for (const [nsid, event] of Object.entries(newEvents)) {

     

       51
       51
       -
                       map.set(nsid, event);

     

       52
       52
       -
                   }

     

       53
       53
       -
                   return map;

     

       54
       54
       -
               });

     

       55
       55
       -
           };

     

       31
       31
       +
           let per_second = $state(0);

     

       32
       32
       +
           let tracking_since = $state(0);

     

       56
       33
        
       

     

       57
       34
        
           let all: EventRecord = $derived(

     

       58
       35
        
               eventsList.reduce(

     
···

       99
       76
        
               };

     

       100
       77
        
               websocket.onmessage = async (event) => {

     

       101
       78
        
                   const jsonData = JSON.parse(event.data);

     

       102
       102
       -
                   per_second = jsonData.per_second;

     

       79
       79
       +
       

     

       80
       80
       +
                   if (jsonData.per_second > 0) {

     

       81
       81
       +
                       per_second = jsonData.per_second;

     

       82
       82
       +
                   }

     

       83
       83
       +
       

     

       84
       84
       +
                   // Store updates in pending map if refresh rate is set

     

       103
       85
        
                   if (refreshRate) {

     

       104
       86
        
                       for (const [nsid, event] of Object.entries(jsonData.events)) {

     

       105
       87
        
                           pendingUpdates.set(nsid, event as EventRecord);

     

       106
       88
        
                       }

     

       107
       89
        
                   } else {

     

       108
       108
       -
                       applyEvents(jsonData.events);

     

       90
       90
       +
                       // Apply updates immediately if no refresh rate

     

       91
       91
       +
                       events.update((map) => {

     

       92
       92
       +
                           for (const [nsid, event] of Object.entries(

     

       93
       93
       +
                               jsonData.events,

     

       94
       94
       +
                           )) {

     

       95
       95
       +
                               map.set(nsid, event as EventRecord);

     

       96
       96
       +
                           }

     

       97
       97
       +
                           return map;

     

       98
       98
       +
                       });

     

       109
       99
        
                   }

     

       110
       100
        
               };

     

       111
       101
        
               websocket.onerror = (error) => {

     
···

       124
       114
        
                   error = null;

     

       125
       115
        
                   const data = await fetchEvents();

     

       126
       116
        
                   per_second = data.per_second;

     

       127
       127
       -
                   applyEvents(data.events);

     

       117
       117
       +
                   events.update((map) => {

     

       118
       118
       +
                       for (const [nsid, event] of Object.entries(data.events)) {

     

       119
       119
       +
                           map.set(nsid, event);

     

       120
       120
       +
                       }

     

       121
       121
       +
                       return map;

     

       122
       122
       +
                   });

     

       128
       123
        
                   tracking_since = (await fetchTrackingSince()).since;

     

       129
       124
        
               } catch (err) {

     

       130
       125
        
                   error =

+28 -42

server/Cargo.lock

···

       18
       18
        
       checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"

     

       19
       19
        
       

     

       20
       20
        
       [[package]]

     

       21
       21
       -
       name = "ahash"

     

       22
       22
       -
       version = "0.8.12"

     

       23
       23
       -
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       24
       24
       -
       checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"

     

       25
       25
       -
       dependencies = [

     

       26
       26
       -
        "cfg-if",

     

       27
       27
       -
        "getrandom 0.3.3",

     

       28
       28
       -
        "once_cell",

     

       29
       29
       -
        "serde",

     

       30
       30
       -
        "version_check",

     

       31
       31
       -
        "zerocopy",

     

       32
       32
       -
       ]

     

       33
       33
       -
       

     

       34
       34
       -
       [[package]]

     

       35
       21
        
       name = "aho-corasick"

     

       36
       22
        
       version = "1.1.3"

     

       37
       23
        
       source = "registry+https://github.com/rust-lang/crates.io-index"

     
···

       60
       46
        
       version = "1.0.98"

     

       61
       47
        
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       62
       48
        
       checksum = "e16d2d3311acee920a9eb8d33b8cbc1787ce4a264e85f964c2404b969bdcd487"

     

       63
       63
       -
       

     

       64
       64
       -
       [[package]]

     

       65
       65
       -
       name = "arc-swap"

     

       66
       66
       -
       version = "1.7.1"

     

       67
       67
       -
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       68
       68
       -
       checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457"

     

       69
       49
        
       

     

       70
       50
        
       [[package]]

     

       71
       51
        
       name = "async-compression"

     
···

       327
       307
        
       version = "0.2.1"

     

       328
       308
        
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       329
       309
        
       checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"

     

       310
       310
       +
       

     

       311
       311
       +
       [[package]]

     

       312
       312
       +
       name = "cmake"

     

       313
       313
       +
       version = "0.1.54"

     

       314
       314
       +
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       315
       315
       +
       checksum = "e7caa3f9de89ddbe2c607f4101924c5abec803763ae9534e4f4d7d8f84aa81f0"

     

       316
       316
       +
       dependencies = [

     

       317
       317
       +
        "cc",

     

       318
       318
       +
       ]

     

       330
       319
        
       

     

       331
       320
        
       [[package]]

     

       332
       321
        
       name = "combine"

     
···

       1563
       1552
        
       name = "server"

     

       1564
       1553
        
       version = "0.1.0"

     

       1565
       1554
        
       dependencies = [

     

       1566
       1566
       -
        "ahash",

     

       1567
       1555
        
        "anyhow",

     

       1568
       1568
       -
        "arc-swap",

     

       1569
       1556
        
        "async-trait",

     

       1570
       1557
        
        "axum",

     

       1571
       1558
        
        "axum-tws",

     
···

       1584
       1571
        
        "serde",

     

       1585
       1572
        
        "serde_json",

     

       1586
       1573
        
        "smol_str",

     

       1574
       1574
       +
        "snmalloc-rs",

     

       1587
       1575
        
        "threadpool",

     

       1588
       1576
        
        "tikv-jemallocator",

     

       1589
       1577
        
        "tokio",

     
···

       1655
       1643
        
       dependencies = [

     

       1656
       1644
        
        "borsh",

     

       1657
       1645
        
        "serde",

     

       1646
       1646
       +
       ]

     

       1647
       1647
       +
       

     

       1648
       1648
       +
       [[package]]

     

       1649
       1649
       +
       name = "snmalloc-rs"

     

       1650
       1650
       +
       version = "0.3.8"

     

       1651
       1651
       +
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       1652
       1652
       +
       checksum = "eb317153089fdfa4d8a2eec059d40a5a23c3bde43995ea23b19121c3f621e74a"

     

       1653
       1653
       +
       dependencies = [

     

       1654
       1654
       +
        "snmalloc-sys",

     

       1655
       1655
       +
       ]

     

       1656
       1656
       +
       

     

       1657
       1657
       +
       [[package]]

     

       1658
       1658
       +
       name = "snmalloc-sys"

     

       1659
       1659
       +
       version = "0.3.8"

     

       1660
       1660
       +
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       1661
       1661
       +
       checksum = "065fea53d32bb77bc36cca466cb191f2e5216ebfd0ed360b1d64889ee6e559ea"

     

       1662
       1662
       +
       dependencies = [

     

       1663
       1663
       +
        "cmake",

     

       1658
       1664
        
       ]

     

       1659
       1665
        
       

     

       1660
       1666
        
       [[package]]

     
···

       2400
       2406
        
       version = "0.8.15"

     

       2401
       2407
        
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       2402
       2408
        
       checksum = "fdd20c5420375476fbd4394763288da7eb0cc0b8c11deed431a91562af7335d3"

     

       2403
       2403
       -
       

     

       2404
       2404
       -
       [[package]]

     

       2405
       2405
       -
       name = "zerocopy"

     

       2406
       2406
       -
       version = "0.8.26"

     

       2407
       2407
       -
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       2408
       2408
       -
       checksum = "1039dd0d3c310cf05de012d8a39ff557cb0d23087fd44cad61df08fc31907a2f"

     

       2409
       2409
       -
       dependencies = [

     

       2410
       2410
       -
        "zerocopy-derive",

     

       2411
       2411
       -
       ]

     

       2412
       2412
       -
       

     

       2413
       2413
       -
       [[package]]

     

       2414
       2414
       -
       name = "zerocopy-derive"

     

       2415
       2415
       -
       version = "0.8.26"

     

       2416
       2416
       -
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       2417
       2417
       -
       checksum = "9ecf5b4cc5364572d7f4c329661bcc82724222973f2cab6f050a4e5c22f75181"

     

       2418
       2418
       -
       dependencies = [

     

       2419
       2419
       -
        "proc-macro2",

     

       2420
       2420
       -
        "quote",

     

       2421
       2421
       -
        "syn",

     

       2422
       2422
       -
       ]

     

       2423
       2409
        
       

     

       2424
       2410
        
       [[package]]

     

       2425
       2411
        
       name = "zeroize"

+2 -2

server/Cargo.toml

···

       30
       30
        
       rayon = "1.10.0"

     

       31
       31
        
       parking_lot = { version = "0.12", features = ["send_guard", "hardware-lock-elision"] }

     

       32
       32
        
       rclite = "0.2.7"

     

       33
       33
       -
       arc-swap = "1.7.1"

     

       34
       34
       -
       ahash = { version = "0.8.12", features = ["serde"] }

     

       35
       33
        
       

     

       34
       34
       +
       [target.'cfg(target_env = "msvc")'.dependencies]

     

       35
       35
       +
       snmalloc-rs = "0.3.8"

     

       36
       36
        
       

     

       37
       37
        
       [target.'cfg(not(target_env = "msvc"))'.dependencies]

     

       38
       38
        
       tikv-jemallocator = "0.6"

+18 -16

server/src/api.rs

···

       1
       1
        
       use std::{

     

       2
       2
       +
           collections::HashMap,

     

       2
       3
        
           fmt::Display,

     

       3
       4
        
           net::SocketAddr,

     

       4
       5
        
           ops::{Bound, Deref, RangeBounds},

     

       5
       6
        
           time::Duration,

     

       6
       7
        
       };

     

       7
       8
        
       

     

       8
       8
       -
       use ahash::AHashMap;

     

       9
       9
        
       use anyhow::anyhow;

     

       10
       10
        
       use axum::{

     

       11
       11
        
           Json, Router,

     
···

       117
       117
        
       #[derive(Serialize)]

     

       118
       118
        
       struct Events {

     

       119
       119
        
           per_second: usize,

     

       120
       120
       -
           events: AHashMap<SmolStr, NsidCount>,

     

       120
       120
       +
           events: HashMap<SmolStr, NsidCount>,

     

       121
       121
        
       }

     

       122
       122
        
       

     

       123
       123
        
       async fn events(db: State<Arc<Db>>) -> AppResult<Json<Events>> {

     

       124
       124
       -
           let mut events = AHashMap::new();

     

       124
       124
       +
           let mut events = HashMap::new();

     

       125
       125
        
           for result in db.get_counts() {

     

       126
       126
        
               let (nsid, counts) = result?;

     

       127
       127
        
               events.insert(

     
···

       176
       176
        
       ) -> AppResult<Json<Vec<Hit>>> {

     

       177
       177
        
           let from = params.to.map(Bound::Included).unwrap_or(Bound::Unbounded);

     

       178
       178
        
           let to = params.from.map(Bound::Included).unwrap_or(Bound::Unbounded);

     

       179
       179
       +
           let maybe_hits = db

     

       180
       180
       +
               .get_hits(&params.nsid, HitsRange { from, to })

     

       181
       181
       +
               .take(MAX_HITS);

     

       182
       182
       +
           let mut hits = Vec::with_capacity(maybe_hits.size_hint().0);

     

       179
       183
        
       

     

       180
       180
       -
           db.get_hits(&params.nsid, HitsRange { from, to }, MAX_HITS)

     

       181
       181
       -
               .take(MAX_HITS)

     

       182
       182
       -
               .try_fold(Vec::with_capacity(MAX_HITS), |mut acc, hit| {

     

       183
       183
       -
                   let hit = hit?;

     

       184
       184
       -
                   let hit_data = hit.deser()?;

     

       184
       184
       +
           for maybe_hit in maybe_hits {

     

       185
       185
       +
               let hit = maybe_hit?;

     

       186
       186
       +
               let hit_data = hit.deser()?;

     

       185
       187
        
       

     

       186
       186
       -
                   acc.push(Hit {

     

       187
       187
       -
                       timestamp: hit.timestamp,

     

       188
       188
       -
                       deleted: hit_data.deleted,

     

       189
       189
       -
                   });

     

       190
       190
       -
                   Ok(acc)

     

       191
       191
       -
               })

     

       192
       192
       -
               .map(Json)

     

       188
       188
       +
               hits.push(Hit {

     

       189
       189
       +
                   timestamp: hit.timestamp,

     

       190
       190
       +
                   deleted: hit_data.deleted,

     

       191
       191
       +
               });

     

       192
       192
       +
           }

     

       193
       193
       +
       

     

       194
       194
       +
           Ok(Json(hits))

     

       193
       195
        
       }

     

       194
       196
        
       

     

       195
       197
        
       async fn stream_events(db: State<Arc<Db>>, ws: WebSocketUpgrade) -> Response {

     
···

       198
       200
        
               (async move {

     

       199
       201
        
                   let mut listener = db.new_listener();

     

       200
       202
        
                   let mut data = Events {

     

       201
       201
       -
                       events: AHashMap::<SmolStr, NsidCount>::with_capacity(10),

     

       203
       203
       +
                       events: HashMap::<SmolStr, NsidCount>::with_capacity(10),

     

       202
       204
        
                       per_second: 0,

     

       203
       205
        
                   };

     

       204
       206
        
                   let mut updates = 0;

+26 -54

server/src/db/handle.rs

···

       1
       1
        
       use std::{

     

       2
       2
        
           fmt::Debug,

     

       3
       3
        
           io::Cursor,

     

       4
       4
       -
           ops::{Bound, RangeBounds},

     

       4
       4
       +
           ops::{Bound, Deref, RangeBounds},

     

       5
       5
        
           sync::atomic::{AtomicU64, Ordering as AtomicOrdering},

     

       6
       6
        
           time::Duration,

     

       7
       7
        
       };

     

       8
       8
        
       

     

       9
       9
        
       use byteview::ByteView;

     

       10
       10
       -
       use fjall::{Keyspace, Partition, PartitionCreateOptions, Slice, Snapshot};

     

       10
       10
       +
       use fjall::{Keyspace, Partition, PartitionCreateOptions, Slice};

     

       11
       11
        
       use itertools::Itertools;

     

       12
       12
        
       use parking_lot::Mutex;

     

       13
       13
        
       use rayon::iter::{IntoParallelIterator, ParallelIterator};

     
···

       16
       16
        
       

     

       17
       17
        
       use crate::{

     

       18
       18
        
           db::{EventRecord, NsidHit, block},

     

       19
       19
       -
           error::{AppError, AppResult},

     

       20
       20
       -
           utils::{

     

       21
       21
       -
               ArcRefCnt, ArcliteSwap, CLOCK, DefaultRateTracker, RateTracker, ReadVariableExt,

     

       22
       22
       -
               varints_unsigned_encoded,

     

       23
       23
       -
           },

     

       19
       19
       +
           error::AppResult,

     

       20
       20
       +
           utils::{CLOCK, DefaultRateTracker, RateTracker, ReadVariableExt, varints_unsigned_encoded},

     

       24
       21
        
       };

     

       25
       22
        
       

     

       26
       23
        
       pub type ItemDecoder = block::ItemDecoder<Cursor<Slice>, NsidHit>;

     
···

       34
       31
        
       }

     

       35
       32
        
       

     

       36
       33
        
       pub struct LexiconHandle {

     

       37
       37
       -
           write_tree: Partition,

     

       38
       38
       -
           read_tree: ArcliteSwap<Snapshot>,

     

       34
       34
       +
           tree: Partition,

     

       39
       35
        
           nsid: SmolStr,

     

       40
       36
        
           buf: Arc<Mutex<Vec<EventRecord>>>,

     

       41
       37
        
           last_insert: AtomicU64, // relaxed

     
···

       50
       46
        
           }

     

       51
       47
        
       }

     

       52
       48
        
       

     

       49
       49
       +
       impl Deref for LexiconHandle {

     

       50
       50
       +
           type Target = Partition;

     

       51
       51
       +
       

     

       52
       52
       +
           fn deref(&self) -> &Self::Target {

     

       53
       53
       +
               &self.tree

     

       54
       54
       +
           }

     

       55
       55
       +
       }

     

       56
       56
       +
       

     

       53
       57
        
       impl LexiconHandle {

     

       54
       58
        
           pub fn new(keyspace: &Keyspace, nsid: &str) -> Self {

     

       55
       59
        
               let opts = PartitionCreateOptions::default()

     

       56
       56
       -
                   .block_size(1024 * 48)

     

       60
       60
       +
                   .block_size(1024 * 128)

     

       57
       61
        
                   .compression(fjall::CompressionType::Miniz(9));

     

       58
       58
       -
               let write_tree = keyspace.open_partition(nsid, opts).unwrap();

     

       59
       59
       -
               let read_tree = ArcliteSwap::new(ArcRefCnt::new(write_tree.snapshot()));

     

       60
       62
        
               Self {

     

       61
       61
       -
                   write_tree,

     

       62
       62
       -
                   read_tree,

     

       63
       63
       +
                   tree: keyspace.open_partition(nsid, opts).unwrap(),

     

       63
       64
        
                   nsid: nsid.into(),

     

       64
       65
        
                   buf: Default::default(),

     

       65
       66
        
                   last_insert: AtomicU64::new(0),

     
···

       67
       68
        
               }

     

       68
       69
        
           }

     

       69
       70
        
       

     

       70
       70
       -
           #[inline(always)]

     

       71
       71
       -
           pub fn read(&self) -> arc_swap::Guard<ArcRefCnt<Snapshot>> {

     

       72
       72
       -
               self.read_tree.load()

     

       73
       73
       -
           }

     

       74
       74
       -
       

     

       75
       75
       -
           #[inline(always)]

     

       76
       76
       -
           pub fn update_tree(&self) {

     

       77
       77
       -
               self.read_tree

     

       78
       78
       -
                   .store(ArcRefCnt::new(self.write_tree.snapshot()));

     

       79
       79
       -
           }

     

       80
       80
       -
       

     

       81
       81
       -
           #[inline(always)]

     

       82
       82
       -
           pub fn span(&self) -> tracing::Span {

     

       83
       83
       -
               tracing::info_span!("handle", nsid = %self.nsid)

     

       84
       84
       -
           }

     

       85
       85
       -
       

     

       86
       86
       -
           #[inline(always)]

     

       87
       71
        
           pub fn nsid(&self) -> &SmolStr {

     

       88
       72
        
               &self.nsid

     

       89
       73
        
           }

     

       90
       74
        
       

     

       91
       91
       -
           #[inline(always)]

     

       92
       75
        
           pub fn item_count(&self) -> usize {

     

       93
       76
        
               self.buf.lock().len()

     

       94
       77
        
           }

     

       95
       78
        
       

     

       96
       96
       -
           pub fn since_last_activity(&self) -> Duration {

     

       97
       97
       -
               Duration::from_nanos(

     

       98
       98
       -
                   CLOCK.delta_as_nanos(self.last_insert.load(AtomicOrdering::Relaxed), CLOCK.raw()),

     

       99
       99
       -
               )

     

       79
       79
       +
           pub fn since_last_activity(&self) -> u64 {

     

       80
       80
       +
               CLOCK.delta_as_nanos(self.last_insert.load(AtomicOrdering::Relaxed), CLOCK.raw())

     

       100
       81
        
           }

     

       101
       82
        
       

     

       102
       83
        
           pub fn suggested_block_size(&self) -> usize {

     
···

       118
       99
        
               range: impl RangeBounds<u64>,

     

       119
       100
        
               sort: bool,

     

       120
       101
        
           ) -> AppResult<()> {

     

       121
       121
       -
               let _span = self.span().entered();

     

       122
       122
       -
       

     

       123
       102
        
               let start_limit = match range.start_bound().cloned() {

     

       124
       103
        
                   Bound::Included(start) => start,

     

       125
       104
        
                   Bound::Excluded(start) => start.saturating_add(1),

     
···

       135
       114
        
               let end_key = varints_unsigned_encoded([end_limit]);

     

       136
       115
        
       

     

       137
       116
        
               let blocks_to_compact = self

     

       138
       138
       -
                   .read()

     

       117
       117
       +
                   .tree

     

       139
       118
        
                   .range(start_key..end_key)

     

       140
       119
        
                   .collect::<Result<Vec<_>, _>>()?;

     

       141
       120
        
               if blocks_to_compact.len() < 2 {

     

       121
       121
       +
                   tracing::info!("{}: nothing to compact", self.nsid);

     

       142
       122
        
                   return Ok(());

     

       143
       123
        
               }

     

       144
       124
        
       

     
···

       175
       155
        
               let end_blocks_size = new_blocks.len();

     

       176
       156
        
       

     

       177
       157
        
               for key in keys_to_delete {

     

       178
       178
       -
                   self.write_tree.remove(key.clone())?;

     

       158
       158
       +
                   self.tree.remove(key.clone())?;

     

       179
       159
        
               }

     

       180
       160
        
               for block in new_blocks {

     

       181
       181
       -
                   self.write_tree.insert(block.key, block.data)?;

     

       161
       161
       +
                   self.tree.insert(block.key, block.data)?;

     

       182
       162
        
               }

     

       183
       163
        
       

     

       184
       184
       -
               let reduction =

     

       185
       185
       -
                   ((start_blocks_size - end_blocks_size) as f64 / start_blocks_size as f64) * 100.0;

     

       186
       164
        
               tracing::info!(

     

       187
       187
       -
                   {

     

       188
       188
       -
                       start = start_blocks_size,

     

       189
       189
       -
                       end = end_blocks_size,

     

       190
       190
       -
                   },

     

       191
       191
       -
                   "blocks compacted {reduction:.2}%",

     

       165
       165
       +
                   "{}: compacted {} blocks to {} blocks ({}% reduction)",

     

       166
       166
       +
                   self.nsid,

     

       167
       167
       +
                   start_blocks_size,

     

       168
       168
       +
                   end_blocks_size,

     

       169
       169
       +
                   ((start_blocks_size - end_blocks_size) as f64 / start_blocks_size as f64) * 100.0,

     

       192
       170
        
               );

     

       193
       171
        
       

     

       194
       172
        
               Ok(())

     

       195
       195
       -
           }

     

       196
       196
       -
       

     

       197
       197
       -
           pub fn insert_block(&self, block: Block) -> AppResult<()> {

     

       198
       198
       -
               self.write_tree

     

       199
       199
       -
                   .insert(block.key, block.data)

     

       200
       200
       -
                   .map_err(AppError::from)

     

       201
       173
        
           }

     

       202
       174
        
       

     

       203
       175
        
           pub fn encode_block_from_items(

+82 -139

server/src/db/mod.rs

···

       1
       1
        
       use std::{

     

       2
       2
       +
           collections::HashMap,

     

       2
       3
        
           fmt::Debug,

     

       3
       4
        
           io::Cursor,

     

       4
       5
        
           ops::{Bound, Deref, RangeBounds},

     

       5
       5
       -
           path::Path,

     

       6
       6
       +
           path::{Path, PathBuf},

     

       6
       7
        
           time::Duration,

     

       7
       7
       -
           u64,

     

       8
       8
        
       };

     

       9
       9
        
       

     

       10
       10
       -
       use ahash::{AHashMap, AHashSet};

     

       11
       10
        
       use byteview::StrView;

     

       12
       12
       -
       use fjall::{Keyspace, Partition, PartitionCreateOptions};

     

       11
       11
       +
       use fjall::{Config, Keyspace, Partition, PartitionCreateOptions};

     

       13
       12
        
       use itertools::{Either, Itertools};

     

       14
       14
       -
       use rayon::iter::{IntoParallelIterator, ParallelIterator};

     

       13
       13
       +
       use rayon::iter::{IndexedParallelIterator, IntoParallelIterator, ParallelIterator};

     

       15
       14
        
       use rclite::Arc;

     

       16
       15
        
       use rkyv::{Archive, Deserialize, Serialize, rancor::Error};

     

       17
       16
        
       use smol_str::{SmolStr, ToSmolStr};

     
···

       22
       21
        
           db::handle::{ItemDecoder, LexiconHandle},

     

       23
       22
        
           error::{AppError, AppResult},

     

       24
       23
        
           jetstream::JetstreamEvent,

     

       25
       25
       -
           utils::{CLOCK, RateTracker, ReadVariableExt, varints_unsigned_encoded},

     

       24
       24
       +
           utils::{RateTracker, ReadVariableExt, varints_unsigned_encoded},

     

       26
       25
        
       };

     

       27
       26
        
       

     

       28
       27
        
       mod block;

     
···

       72
       71
        
       }

     

       73
       72
        
       

     

       74
       73
        
       pub struct DbInfo {

     

       75
       75
       -
           pub nsids: AHashMap<SmolStr, Vec<usize>>,

     

       74
       74
       +
           pub nsids: HashMap<SmolStr, Vec<usize>>,

     

       76
       75
        
           pub disk_size: u64,

     

       77
       76
        
       }

     

       78
       77
        
       

     
···

       80
       79
        
           pub ks_config: fjall::Config,

     

       81
       80
        
           pub min_block_size: usize,

     

       82
       81
        
           pub max_block_size: usize,

     

       83
       83
       -
           pub max_last_activity: Duration,

     

       82
       82
       +
           pub max_last_activity: u64,

     

       84
       83
        
       }

     

       85
       84
        
       

     

       86
       85
        
       impl DbConfig {

     
···

       98
       97
        
       impl Default for DbConfig {

     

       99
       98
        
           fn default() -> Self {

     

       100
       99
        
               Self {

     

       101
       101
       -
                   ks_config: fjall::Config::default()

     

       102
       102
       -
                       .cache_size(1024 * 1024 * 512)

     

       103
       103
       -
                       .max_write_buffer_size(u64::MAX),

     

       104
       104
       -
                   min_block_size: 1000,

     

       105
       105
       -
                   max_block_size: 250_000,

     

       106
       106
       -
                   max_last_activity: Duration::from_secs(10),

     

       100
       100
       +
                   ks_config: fjall::Config::default(),

     

       101
       101
       +
                   min_block_size: 512,

     

       102
       102
       +
                   max_block_size: 500_000,

     

       103
       103
       +
                   max_last_activity: Duration::from_secs(10).as_nanos() as u64,

     

       107
       104
        
               }

     

       108
       105
        
           }

     

       109
       106
        
       }

     
···

       114
       111
        
           pub cfg: DbConfig,

     

       115
       112
        
           pub ks: Keyspace,

     

       116
       113
        
           counts: Partition,

     

       117
       117
       -
           hits: scc::HashIndex<SmolStr, Arc<LexiconHandle>, ahash::RandomState>,

     

       114
       114
       +
           hits: scc::HashIndex<SmolStr, Arc<LexiconHandle>>,

     

       118
       115
        
           sync_pool: threadpool::ThreadPool,

     

       119
       116
        
           event_broadcaster: broadcast::Sender<(SmolStr, NsidCounts)>,

     

       120
       120
       -
           eps: RateTracker<100>, // 100 millis buckets

     

       117
       117
       +
           eps: RateTracker<100>,

     

       121
       118
        
           cancel_token: CancellationToken,

     

       122
       119
        
       }

     

       123
       120
        
       

     
···

       163
       160
        
           }

     

       164
       161
        
       

     

       165
       162
        
           pub fn sync(&self, all: bool) -> AppResult<()> {

     

       166
       166
       -
               let start = CLOCK.now();

     

       167
       163
        
               // prepare all the data

     

       168
       168
       -
               let nsids_len = self.hits.len();

     

       169
       169
       -
               let mut data = Vec::with_capacity(nsids_len);

     

       170
       170
       -
               let mut nsids = AHashSet::with_capacity(nsids_len);

     

       164
       164
       +
               let mut data = Vec::with_capacity(self.hits.len());

     

       171
       165
        
               let _guard = scc::ebr::Guard::new();

     

       172
       172
       -
               for (nsid, handle) in self.hits.iter(&_guard) {

     

       166
       166
       +
               for (_, handle) in self.hits.iter(&_guard) {

     

       173
       167
        
                   let mut nsid_data = Vec::with_capacity(2);

     

       174
       174
       -
                   // let mut total_count = 0;

     

       168
       168
       +
                   let mut total_count = 0;

     

       175
       169
        
                   let is_too_old = handle.since_last_activity() > self.cfg.max_last_activity;

     

       176
       170
        
                   // if we disconnect for a long time, we want to sync all of what we

     

       177
       171
        
                   // have to avoid having many small blocks (even if we run compaction

     
···

       186
       180
        
                   let count = handle.item_count();

     

       187
       181
        
                   let data_count = count / block_size;

     

       188
       182
        
                   if count > 0 && (all || data_count > 0 || is_too_old) {

     

       189
       189
       -
                       for _ in 0..data_count {

     

       190
       190
       -
                           nsid_data.push((handle.clone(), block_size));

     

       191
       191
       -
                           // total_count += block_size;

     

       183
       183
       +
                       for i in 0..data_count {

     

       184
       184
       +
                           nsid_data.push((i, handle.clone(), block_size));

     

       185
       185
       +
                           total_count += block_size;

     

       192
       186
        
                       }

     

       193
       187
        
                       // only sync remainder if we haven't met block size

     

       194
       188
        
                       let remainder = count % block_size;

     

       195
       189
        
                       if (all || data_count == 0) && remainder > 0 {

     

       196
       196
       -
                           nsid_data.push((handle.clone(), remainder));

     

       197
       197
       -
                           // total_count += remainder;

     

       190
       190
       +
                           nsid_data.push((data_count, handle.clone(), remainder));

     

       191
       191
       +
                           total_count += remainder;

     

       198
       192
        
                       }

     

       199
       193
        
                   }

     

       200
       200
       -
                   let _span = handle.span().entered();

     

       201
       201
       -
                   if nsid_data.len() > 0 {

     

       202
       202
       -
                       // tracing::info!(

     

       203
       203
       -
                       //     {blocks = %nsid_data.len(), count = %total_count},

     

       204
       204
       -
                       //     "will encode & sync",

     

       205
       205
       -
                       // );

     

       206
       206
       -
                       nsids.insert(nsid.clone());

     

       207
       207
       -
                       data.push(nsid_data);

     

       208
       208
       -
                   }

     

       194
       194
       +
                   tracing::info!(

     

       195
       195
       +
                       "{}: will sync {} blocks ({} count)",

     

       196
       196
       +
                       handle.nsid(),

     

       197
       197
       +
                       nsid_data.len(),

     

       198
       198
       +
                       total_count,

     

       199
       199
       +
                   );

     

       200
       200
       +
                   data.push(nsid_data);

     

       209
       201
        
               }

     

       210
       202
        
               drop(_guard);

     

       211
       203
        
       

     
···

       214
       206
        
                   .map(|chunk| {

     

       215
       207
        
                       chunk

     

       216
       208
        
                           .into_iter()

     

       217
       217
       -
                           .map(|(handle, max_block_size)| {

     

       218
       218
       -
                               (handle.take_block_items(max_block_size), handle)

     

       209
       209
       +
                           .map(|(i, handle, max_block_size)| {

     

       210
       210
       +
                               (i, handle.take_block_items(max_block_size), handle)

     

       219
       211
        
                           })

     

       220
       212
        
                           .collect::<Vec<_>>()

     

       221
       213
        
                           .into_par_iter()

     

       222
       222
       -
                           .map(|(items, handle)| {

     

       214
       214
       +
                           .map(|(i, items, handle)| {

     

       223
       215
        
                               let count = items.len();

     

       224
       216
        
                               let block = LexiconHandle::encode_block_from_items(items, count)?;

     

       225
       225
       -
                               AppResult::Ok((block, handle))

     

       217
       217
       +
                               tracing::info!(

     

       218
       218
       +
                                   "{}: encoded block with {} items",

     

       219
       219
       +
                                   handle.nsid(),

     

       220
       220
       +
                                   block.written,

     

       221
       221
       +
                               );

     

       222
       222
       +
                               AppResult::Ok((i, block, handle))

     

       226
       223
        
                           })

     

       227
       224
        
                           .collect::<Result<Vec<_>, _>>()

     

       228
       225
        
                   })

     

       229
       226
        
                   .try_for_each(|chunk| {

     

       230
       227
        
                       let chunk = chunk?;

     

       231
       231
       -
                       for (block, handle) in chunk {

     

       232
       232
       -
                           self.sync_pool.execute(move || {

     

       233
       233
       -
                               let _span = handle.span().entered();

     

       234
       234
       -
                               let written = block.written;

     

       235
       235
       -
                               match handle.insert_block(block) {

     

       228
       228
       +
                       for (i, block, handle) in chunk {

     

       229
       229
       +
                           self.sync_pool

     

       230
       230
       +
                               .execute(move || match handle.insert(block.key, block.data) {

     

       236
       231
        
                                   Ok(_) => {

     

       237
       237
       -
                                       tracing::info!({count = %written}, "synced")

     

       232
       232
       +
                                       tracing::info!("{}: [{i}] synced {}", block.written, handle.nsid())

     

       238
       233
        
                                   }

     

       239
       239
       -
                                   Err(err) => tracing::error!({ err = %err }, "failed to sync block"),

     

       240
       240
       -
                               }

     

       241
       241
       -
                           });

     

       234
       234
       +
                                   Err(err) => tracing::error!("failed to sync block: {}", err),

     

       235
       235
       +
                               });

     

       242
       236
        
                       }

     

       243
       237
        
                       AppResult::Ok(())

     

       244
       238
        
                   })?;

     

       245
       239
        
               self.sync_pool.join();

     

       246
       240
        
       

     

       247
       247
       -
               // update snapshots for all (changed) handles

     

       248
       248
       -
               for nsid in nsids {

     

       249
       249
       -
                   self.hits.peek_with(&nsid, |_, handle| handle.update_tree());

     

       250
       250
       -
               }

     

       251
       251
       -
       

     

       252
       252
       -
               tracing::info!(time = %start.elapsed().as_secs_f64(), "synced all blocks");

     

       253
       253
       -
       

     

       254
       241
        
               Ok(())

     

       255
       242
        
           }

     

       256
       243
        
       

     
···

       264
       251
        
               let Some(handle) = self.get_handle(nsid) else {

     

       265
       252
        
                   return Ok(());

     

       266
       253
        
               };

     

       267
       267
       -
               handle.compact(max_count, range, sort)?;

     

       268
       268
       -
               handle.update_tree();

     

       269
       269
       -
               Ok(())

     

       254
       254
       +
               handle.compact(max_count, range, sort)

     

       270
       255
        
           }

     

       271
       256
        
       

     

       272
       257
        
           pub fn compact_all(

     
···

       283
       268
        
       

     

       284
       269
        
           pub fn major_compact(&self) -> AppResult<()> {

     

       285
       270
        
               self.compact_all(self.cfg.max_block_size, .., true)?;

     

       271
       271
       +
               let _guard = scc::ebr::Guard::new();

     

       272
       272
       +
               for (_, handle) in self.hits.iter(&_guard) {

     

       273
       273
       +
                   handle.deref().major_compact()?;

     

       274
       274
       +
               }

     

       286
       275
        
               Ok(())

     

       287
       276
        
           }

     

       288
       277
        
       

     
···

       312
       301
        
           }

     

       313
       302
        
       

     

       314
       303
        
           pub fn ingest_events(&self, events: impl Iterator<Item = EventRecord>) -> AppResult<()> {

     

       315
       315
       -
               let mut seen_events = 0;

     

       316
       304
        
               for (key, chunk) in events.chunk_by(|event| event.nsid.clone()).into_iter() {

     

       317
       305
        
                   let mut counts = self.get_count(&key)?;

     

       306
       306
       +
                   let mut count = 0;

     

       318
       307
        
                   self.ensure_handle(&key).queue(chunk.inspect(|e| {

     

       319
       308
        
                       // increment count

     

       320
       309
        
                       counts.last_seen = e.timestamp;

     
···

       323
       312
        
                       } else {

     

       324
       313
        
                           counts.count += 1;

     

       325
       314
        
                       }

     

       326
       326
       -
                       seen_events += 1;

     

       315
       315
       +
                       count += 1;

     

       327
       316
        
                   }));

     

       317
       317
       +
                   self.eps.observe(count);

     

       328
       318
        
                   self.insert_count(&key, &counts)?;

     

       329
       319
        
                   if self.event_broadcaster.receiver_count() > 0 {

     

       330
       320
        
                       let _ = self.event_broadcaster.send((key, counts));

     

       331
       321
        
                   }

     

       332
       322
        
               }

     

       333
       333
       -
               self.eps.observe(seen_events);

     

       334
       323
        
               Ok(())

     

       335
       324
        
           }

     

       336
       325
        
       

     
···

       370
       359
        
           }

     

       371
       360
        
       

     

       372
       361
        
           pub fn info(&self) -> AppResult<DbInfo> {

     

       373
       373
       -
               let mut nsids = AHashMap::new();

     

       362
       362
       +
               let mut nsids = HashMap::new();

     

       374
       363
        
               for nsid in self.get_nsids() {

     

       375
       364
        
                   let Some(handle) = self.get_handle(&nsid) else {

     

       376
       365
        
                       continue;

     

       377
       366
        
                   };

     

       378
       378
       -
                   let block_lens = handle

     

       379
       379
       -
                       .read()

     

       380
       380
       -
                       .iter()

     

       381
       381
       -
                       .rev()

     

       382
       382
       -
                       .try_fold(Vec::new(), |mut acc, item| {

     

       383
       383
       -
                           let (key, value) = item?;

     

       384
       384
       -
                           let mut timestamps = Cursor::new(key);

     

       385
       385
       -
                           let start_timestamp = timestamps.read_varint()?;

     

       386
       386
       -
                           let decoder = ItemDecoder::new(Cursor::new(value), start_timestamp)?;

     

       387
       387
       -
                           acc.push(decoder.item_count());

     

       388
       388
       -
                           AppResult::Ok(acc)

     

       389
       389
       -
                       })?;

     

       367
       367
       +
                   let block_lens = handle.iter().rev().try_fold(Vec::new(), |mut acc, item| {

     

       368
       368
       +
                       let (key, value) = item?;

     

       369
       369
       +
                       let mut timestamps = Cursor::new(key);

     

       370
       370
       +
                       let start_timestamp = timestamps.read_varint()?;

     

       371
       371
       +
                       let decoder = ItemDecoder::new(Cursor::new(value), start_timestamp)?;

     

       372
       372
       +
                       acc.push(decoder.item_count());

     

       373
       373
       +
                       AppResult::Ok(acc)

     

       374
       374
       +
                   })?;

     

       390
       375
        
                   nsids.insert(nsid.to_smolstr(), block_lens);

     

       391
       376
        
               }

     

       392
       377
        
               Ok(DbInfo {

     
···

       399
       384
        
               &self,

     

       400
       385
        
               nsid: &str,

     

       401
       386
        
               range: impl RangeBounds<u64> + std::fmt::Debug,

     

       402
       402
       -
               max_items: usize,

     

       403
       387
        
           ) -> impl Iterator<Item = AppResult<handle::Item>> {

     

       404
       388
        
               let start_limit = match range.start_bound().cloned() {

     

       405
       389
        
                   Bound::Included(start) => start,

     
···

       417
       401
        
                   return Either::Right(std::iter::empty());

     

       418
       402
        
               };

     

       419
       403
        
       

     

       420
       420
       -
               // let mut ts = CLOCK.now();

     

       421
       421
       -
               let map_block = move |(res, current_item_count)| -> AppResult<(Option<_>, usize)> {

     

       422
       422
       -
                   if current_item_count >= max_items {

     

       423
       423
       -
                       return Ok((None, current_item_count));

     

       424
       424
       -
                   }

     

       425
       425
       -
                   let (key, val) = res?;

     

       404
       404
       +
               let map_block = move |(key, val)| {

     

       426
       405
        
                   let mut key_reader = Cursor::new(key);

     

       427
       406
        
                   let start_timestamp = key_reader.read_varint::<u64>()?;

     

       428
       428
       -
                   // let end_timestamp = key_reader.read_varint::<u64>()?;

     

       429
       407
        
                   if start_timestamp < start_limit {

     

       430
       430
       -
                       // tracing::info!(

     

       431
       431
       -
                       //     "stopped at block with timestamps {start_timestamp}..{end_timestamp} because {start_limit} is greater"

     

       432
       432
       -
                       // );

     

       433
       433
       -
                       return Ok((None, current_item_count));

     

       408
       408
       +
                       return Ok(None);

     

       434
       409
        
                   }

     

       435
       435
       -
                   let decoder = handle::ItemDecoder::new(Cursor::new(val), start_timestamp)?;

     

       436
       436
       -
                   let current_item_count = current_item_count + decoder.item_count();

     

       437
       437
       -
                   // tracing::info!(

     

       438
       438
       -
                   //     "took {}ns to get block with size {}",

     

       439
       439
       -
                   //     ts.elapsed().as_nanos(),

     

       440
       440
       -
                   //     decoder.item_count()

     

       441
       441
       -
                   // );

     

       442
       442
       -
                   // ts = CLOCK.now();

     

       443
       443
       -
                   Ok((

     

       444
       444
       -
                       Some(

     

       445
       445
       -
                           decoder

     

       446
       446
       -
                               .take_while(move |item| {

     

       447
       447
       -
                                   item.as_ref().map_or(true, |item| {

     

       448
       448
       -
                                       item.timestamp <= end_limit && item.timestamp >= start_limit

     

       449
       449
       -
                                   })

     

       450
       450
       -
                               })

     

       451
       451
       -
                               .map(|res| res.map_err(AppError::from)),

     

       452
       452
       -
                       ),

     

       453
       453
       -
                       current_item_count,

     

       454
       454
       -
                   ))

     

       410
       410
       +
                   let items = handle::ItemDecoder::new(Cursor::new(val), start_timestamp)?

     

       411
       411
       +
                       .take_while(move |item| {

     

       412
       412
       +
                           item.as_ref().map_or(true, |item| {

     

       413
       413
       +
                               item.timestamp <= end_limit && item.timestamp >= start_limit

     

       414
       414
       +
                           })

     

       415
       415
       +
                       })

     

       416
       416
       +
                       .map(|res| res.map_err(AppError::from));

     

       417
       417
       +
                   Ok(Some(items))

     

       455
       418
        
               };

     

       456
       419
        
       

     

       457
       457
       -
               let (blocks, _counted) = handle

     

       458
       458
       -
                   .read()

     

       459
       459
       -
                   .range(..end_key)

     

       460
       460
       -
                   .map(|res| res.map_err(AppError::from))

     

       461
       461
       -
                   .rev()

     

       462
       462
       -
                   .fold_while(

     

       463
       463
       -
                       (Vec::with_capacity(20), 0),

     

       464
       464
       -
                       |(mut blocks, current_item_count), res| {

     

       465
       465
       -
                           use itertools::FoldWhile::*;

     

       466
       466
       -
       

     

       467
       467
       -
                           match map_block((res, current_item_count)) {

     

       468
       468
       -
                               Ok((Some(block), current_item_count)) => {

     

       469
       469
       -
                                   blocks.push(Ok(block));

     

       470
       470
       -
                                   Continue((blocks, current_item_count))

     

       471
       471
       -
                               }

     

       472
       472
       -
                               Ok((None, current_item_count)) => Done((blocks, current_item_count)),

     

       473
       473
       -
                               Err(err) => {

     

       474
       474
       -
                                   blocks.push(Err(err));

     

       475
       475
       -
                                   Done((blocks, current_item_count))

     

       476
       476
       -
                               }

     

       477
       477
       -
                           }

     

       478
       478
       -
                       },

     

       479
       479
       -
                   )

     

       480
       480
       -
                   .into_inner();

     

       481
       481
       -
       

     

       482
       482
       -
               // tracing::info!(

     

       483
       483
       -
               //     "got blocks with size {}, item count {counted}",

     

       484
       484
       -
               //     blocks.len()

     

       485
       485
       -
               // );

     

       486
       486
       -
       

     

       487
       487
       -
               Either::Left(blocks.into_iter().rev().flatten().flatten())

     

       420
       420
       +
               Either::Left(

     

       421
       421
       +
                   handle

     

       422
       422
       +
                       .range(..end_key)

     

       423
       423
       +
                       .rev()

     

       424
       424
       +
                       .map_while(move |res| res.map_err(AppError::from).and_then(map_block).transpose())

     

       425
       425
       +
                       .collect::<Vec<_>>()

     

       426
       426
       +
                       .into_iter()

     

       427
       427
       +
                       .rev()

     

       428
       428
       +
                       .flatten()

     

       429
       429
       +
                       .flatten(),

     

       430
       430
       +
               )

     

       488
       431
        
           }

     

       489
       432
        
       

     

       490
       433
        
           pub fn tracking_since(&self) -> AppResult<u64> {

     
···

       493
       436
        
               let Some(handle) = self.get_handle("app.bsky.feed.like") else {

     

       494
       437
        
                   return Ok(0);

     

       495
       438
        
               };

     

       496
       496
       -
               let Some((timestamps_raw, _)) = handle.read().first_key_value()? else {

     

       439
       439
       +
               let Some((timestamps_raw, _)) = handle.first_key_value()? else {

     

       497
       440
        
                   return Ok(0);

     

       498
       441
        
               };

     

       499
       442
        
               let mut timestamp_reader = Cursor::new(timestamps_raw);

+501

server/src/db_old/block.rs

···

       1
       1
       +
       use ordered_varint::Variable;

     

       2
       2
       +
       use rkyv::{

     

       3
       3
       +
           Archive, Deserialize, Serialize,

     

       4
       4
       +
           api::high::{HighSerializer, HighValidator},

     

       5
       5
       +
           bytecheck::CheckBytes,

     

       6
       6
       +
           de::Pool,

     

       7
       7
       +
           rancor::{self, Strategy},

     

       8
       8
       +
           ser::allocator::ArenaHandle,

     

       9
       9
       +
           util::AlignedVec,

     

       10
       10
       +
       };

     

       11
       11
       +
       use std::{

     

       12
       12
       +
           io::{self, Read, Write},

     

       13
       13
       +
           marker::PhantomData,

     

       14
       14
       +
       };

     

       15
       15
       +
       

     

       16
       16
       +
       use crate::error::{AppError, AppResult};

     

       17
       17
       +
       

     

       18
       18
       +
       pub struct Item<T> {

     

       19
       19
       +
           pub timestamp: u64,

     

       20
       20
       +
           data: AlignedVec,

     

       21
       21
       +
           phantom: PhantomData<T>,

     

       22
       22
       +
       }

     

       23
       23
       +
       

     

       24
       24
       +
       impl<T: Archive> Item<T> {

     

       25
       25
       +
           pub fn access(&self) -> &T::Archived {

     

       26
       26
       +
               unsafe { rkyv::access_unchecked::<T::Archived>(&self.data) }

     

       27
       27
       +
           }

     

       28
       28
       +
       }

     

       29
       29
       +
       

     

       30
       30
       +
       impl<T> Item<T>

     

       31
       31
       +
       where

     

       32
       32
       +
           T: Archive,

     

       33
       33
       +
           T::Archived: for<'a> CheckBytes<HighValidator<'a, rancor::Error>>

     

       34
       34
       +
               + Deserialize<T, Strategy<Pool, rancor::Error>>,

     

       35
       35
       +
       {

     

       36
       36
       +
           pub fn deser(&self) -> AppResult<T> {

     

       37
       37
       +
               rkyv::from_bytes(&self.data).map_err(AppError::from)

     

       38
       38
       +
           }

     

       39
       39
       +
       }

     

       40
       40
       +
       

     

       41
       41
       +
       impl<T: for<'a> Serialize<HighSerializer<AlignedVec, ArenaHandle<'a>, rancor::Error>>> Item<T> {

     

       42
       42
       +
           pub fn new(timestamp: u64, data: &T) -> Self {

     

       43
       43
       +
               Item {

     

       44
       44
       +
                   timestamp,

     

       45
       45
       +
                   data: unsafe { rkyv::to_bytes(data).unwrap_unchecked() },

     

       46
       46
       +
                   phantom: PhantomData,

     

       47
       47
       +
               }

     

       48
       48
       +
           }

     

       49
       49
       +
       }

     

       50
       50
       +
       

     

       51
       51
       +
       pub struct ItemEncoder<W: Write, T> {

     

       52
       52
       +
           writer: W,

     

       53
       53
       +
           prev_timestamp: u64,

     

       54
       54
       +
           prev_delta: i64,

     

       55
       55
       +
           _item: PhantomData<T>,

     

       56
       56
       +
       }

     

       57
       57
       +
       

     

       58
       58
       +
       impl<W: Write, T> ItemEncoder<W, T> {

     

       59
       59
       +
           pub fn new(writer: W) -> Self {

     

       60
       60
       +
               ItemEncoder {

     

       61
       61
       +
                   writer,

     

       62
       62
       +
                   prev_timestamp: 0,

     

       63
       63
       +
                   prev_delta: 0,

     

       64
       64
       +
                   _item: PhantomData,

     

       65
       65
       +
               }

     

       66
       66
       +
           }

     

       67
       67
       +
       

     

       68
       68
       +
           pub fn encode(&mut self, item: &Item<T>) -> AppResult<()> {

     

       69
       69
       +
               if self.prev_timestamp == 0 {

     

       70
       70
       +
                   // self.writer.write_varint(item.timestamp)?;

     

       71
       71
       +
                   self.prev_timestamp = item.timestamp;

     

       72
       72
       +
                   self.write_data(&item.data)?;

     

       73
       73
       +
                   return Ok(());

     

       74
       74
       +
               }

     

       75
       75
       +
       

     

       76
       76
       +
               let delta = (item.timestamp as i128 - self.prev_timestamp as i128) as i64;

     

       77
       77
       +
       

     

       78
       78
       +
               self.writer.write_varint(delta - self.prev_delta)?;

     

       79
       79
       +
               self.prev_timestamp = item.timestamp;

     

       80
       80
       +
               self.prev_delta = delta;

     

       81
       81
       +
       

     

       82
       82
       +
               self.write_data(&item.data)?;

     

       83
       83
       +
       

     

       84
       84
       +
               Ok(())

     

       85
       85
       +
           }

     

       86
       86
       +
       

     

       87
       87
       +
           fn write_data(&mut self, data: &[u8]) -> AppResult<()> {

     

       88
       88
       +
               self.writer.write_varint(data.len())?;

     

       89
       89
       +
               self.writer.write_all(data)?;

     

       90
       90
       +
               Ok(())

     

       91
       91
       +
           }

     

       92
       92
       +
       

     

       93
       93
       +
           pub fn finish(mut self) -> AppResult<W> {

     

       94
       94
       +
               self.writer.flush()?;

     

       95
       95
       +
               Ok(self.writer)

     

       96
       96
       +
           }

     

       97
       97
       +
       }

     

       98
       98
       +
       

     

       99
       99
       +
       pub struct ItemDecoder<R, T> {

     

       100
       100
       +
           reader: R,

     

       101
       101
       +
           current_timestamp: u64,

     

       102
       102
       +
           current_delta: i64,

     

       103
       103
       +
           first_item: bool,

     

       104
       104
       +
           _item: PhantomData<T>,

     

       105
       105
       +
       }

     

       106
       106
       +
       

     

       107
       107
       +
       impl<R: Read, T: Archive> ItemDecoder<R, T> {

     

       108
       108
       +
           pub fn new(reader: R, start_timestamp: u64) -> AppResult<Self> {

     

       109
       109
       +
               Ok(ItemDecoder {

     

       110
       110
       +
                   reader,

     

       111
       111
       +
                   current_timestamp: start_timestamp,

     

       112
       112
       +
                   current_delta: 0,

     

       113
       113
       +
                   first_item: true,

     

       114
       114
       +
                   _item: PhantomData,

     

       115
       115
       +
               })

     

       116
       116
       +
           }

     

       117
       117
       +
       

     

       118
       118
       +
           pub fn decode(&mut self) -> AppResult<Option<Item<T>>> {

     

       119
       119
       +
               if self.first_item {

     

       120
       120
       +
                   // read the first timestamp

     

       121
       121
       +
                   // let timestamp = match self.reader.read_varint::<u64>() {

     

       122
       122
       +
                   //     Ok(timestamp) => timestamp,

     

       123
       123
       +
                   //     Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),

     

       124
       124
       +
                   //     Err(e) => return Err(e.into()),

     

       125
       125
       +
                   // };

     

       126
       126
       +
                   // self.current_timestamp = timestamp;

     

       127
       127
       +
       

     

       128
       128
       +
                   let Some(data_raw) = self.read_item()? else {

     

       129
       129
       +
                       return Ok(None);

     

       130
       130
       +
                   };

     

       131
       131
       +
                   self.first_item = false;

     

       132
       132
       +
                   return Ok(Some(Item {

     

       133
       133
       +
                       timestamp: self.current_timestamp,

     

       134
       134
       +
                       data: data_raw,

     

       135
       135
       +
                       phantom: PhantomData,

     

       136
       136
       +
                   }));

     

       137
       137
       +
               }

     

       138
       138
       +
       

     

       139
       139
       +
               let Some(_delta) = self.read_timestamp()? else {

     

       140
       140
       +
                   return Ok(None);

     

       141
       141
       +
               };

     

       142
       142
       +
       

     

       143
       143
       +
               // read data

     

       144
       144
       +
               let data_raw = match self.read_item()? {

     

       145
       145
       +
                   Some(data_raw) => data_raw,

     

       146
       146
       +
                   None => {

     

       147
       147
       +
                       return Err(io::Error::new(

     

       148
       148
       +
                           io::ErrorKind::UnexpectedEof,

     

       149
       149
       +
                           "expected data after delta",

     

       150
       150
       +
                       )

     

       151
       151
       +
                       .into());

     

       152
       152
       +
                   }

     

       153
       153
       +
               };

     

       154
       154
       +
       

     

       155
       155
       +
               Ok(Some(Item {

     

       156
       156
       +
                   timestamp: self.current_timestamp,

     

       157
       157
       +
                   data: data_raw,

     

       158
       158
       +
                   phantom: PhantomData,

     

       159
       159
       +
               }))

     

       160
       160
       +
           }

     

       161
       161
       +
       

     

       162
       162
       +
           // [10, 11, 12, 14] -> [1, 1, 2] -> [0, 1]

     

       163
       163
       +
           fn read_timestamp(&mut self) -> AppResult<Option<u64>> {

     

       164
       164
       +
               let delta = match self.reader.read_varint::<i64>() {

     

       165
       165
       +
                   Ok(delta) => delta,

     

       166
       166
       +
                   Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),

     

       167
       167
       +
                   Err(e) => return Err(e.into()),

     

       168
       168
       +
               };

     

       169
       169
       +
               self.current_delta += delta;

     

       170
       170
       +
               self.current_timestamp =

     

       171
       171
       +
                   (self.current_timestamp as i128 + self.current_delta as i128) as u64;

     

       172
       172
       +
               Ok(Some(self.current_timestamp))

     

       173
       173
       +
           }

     

       174
       174
       +
       

     

       175
       175
       +
           fn read_item(&mut self) -> AppResult<Option<AlignedVec>> {

     

       176
       176
       +
               let data_len = match self.reader.read_varint::<usize>() {

     

       177
       177
       +
                   Ok(data_len) => data_len,

     

       178
       178
       +
                   Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),

     

       179
       179
       +
                   Err(e) => return Err(e.into()),

     

       180
       180
       +
               };

     

       181
       181
       +
               let mut data_raw = AlignedVec::with_capacity(data_len);

     

       182
       182
       +
               for _ in 0..data_len {

     

       183
       183
       +
                   data_raw.push(0);

     

       184
       184
       +
               }

     

       185
       185
       +
               self.reader.read_exact(data_raw.as_mut_slice())?;

     

       186
       186
       +
               Ok(Some(data_raw))

     

       187
       187
       +
           }

     

       188
       188
       +
       }

     

       189
       189
       +
       

     

       190
       190
       +
       impl<R: Read, T: Archive> Iterator for ItemDecoder<R, T> {

     

       191
       191
       +
           type Item = AppResult<Item<T>>;

     

       192
       192
       +
       

     

       193
       193
       +
           fn next(&mut self) -> Option<Self::Item> {

     

       194
       194
       +
               self.decode().transpose()

     

       195
       195
       +
           }

     

       196
       196
       +
       }

     

       197
       197
       +
       

     

       198
       198
       +
       pub trait WriteVariableExt: Write {

     

       199
       199
       +
           fn write_varint(&mut self, value: impl Variable) -> io::Result<usize> {

     

       200
       200
       +
               value.encode_variable(self)

     

       201
       201
       +
           }

     

       202
       202
       +
       }

     

       203
       203
       +
       impl<W: Write> WriteVariableExt for W {}

     

       204
       204
       +
       

     

       205
       205
       +
       pub trait ReadVariableExt: Read {

     

       206
       206
       +
           fn read_varint<T: Variable>(&mut self) -> io::Result<T> {

     

       207
       207
       +
               T::decode_variable(self)

     

       208
       208
       +
           }

     

       209
       209
       +
       }

     

       210
       210
       +
       impl<R: Read> ReadVariableExt for R {}

     

       211
       211
       +
       

     

       212
       212
       +
       #[cfg(test)]

     

       213
       213
       +
       mod test {

     

       214
       214
       +
           use super::*;

     

       215
       215
       +
           use rkyv::{Archive, Deserialize, Serialize};

     

       216
       216
       +
           use std::io::Cursor;

     

       217
       217
       +
       

     

       218
       218
       +
           #[derive(Archive, Deserialize, Serialize, Debug, PartialEq)]

     

       219
       219
       +
           #[rkyv(compare(PartialEq))]

     

       220
       220
       +
           struct TestData {

     

       221
       221
       +
               id: u32,

     

       222
       222
       +
               value: String,

     

       223
       223
       +
           }

     

       224
       224
       +
       

     

       225
       225
       +
           #[test]

     

       226
       226
       +
           fn test_encoder_decoder_single_item() {

     

       227
       227
       +
               let data = TestData {

     

       228
       228
       +
                   id: 123,

     

       229
       229
       +
                   value: "test".to_string(),

     

       230
       230
       +
               };

     

       231
       231
       +
       

     

       232
       232
       +
               let item = Item::new(1000, &data);

     

       233
       233
       +
       

     

       234
       234
       +
               // encode

     

       235
       235
       +
               let mut buffer = Vec::new();

     

       236
       236
       +
               let mut encoder = ItemEncoder::new(&mut buffer);

     

       237
       237
       +
               encoder.encode(&item).unwrap();

     

       238
       238
       +
               encoder.finish().unwrap();

     

       239
       239
       +
       

     

       240
       240
       +
               // decode

     

       241
       241
       +
               let cursor = Cursor::new(buffer);

     

       242
       242
       +
               let mut decoder = ItemDecoder::<_, TestData>::new(cursor, 1000).unwrap();

     

       243
       243
       +
       

     

       244
       244
       +
               let decoded_item = decoder.decode().unwrap().unwrap();

     

       245
       245
       +
               assert_eq!(decoded_item.timestamp, 1000);

     

       246
       246
       +
       

     

       247
       247
       +
               let decoded_data = decoded_item.access();

     

       248
       248
       +
               assert_eq!(decoded_data.id, 123);

     

       249
       249
       +
               assert_eq!(decoded_data.value.as_str(), "test");

     

       250
       250
       +
           }

     

       251
       251
       +
       

     

       252
       252
       +
           #[test]

     

       253
       253
       +
           fn test_encoder_decoder_multiple_items() {

     

       254
       254
       +
               let items = vec![

     

       255
       255
       +
                   Item::new(

     

       256
       256
       +
                       1000,

     

       257
       257
       +
                       &TestData {

     

       258
       258
       +
                           id: 1,

     

       259
       259
       +
                           value: "first".to_string(),

     

       260
       260
       +
                       },

     

       261
       261
       +
                   ),

     

       262
       262
       +
                   Item::new(

     

       263
       263
       +
                       1010,

     

       264
       264
       +
                       &TestData {

     

       265
       265
       +
                           id: 2,

     

       266
       266
       +
                           value: "second".to_string(),

     

       267
       267
       +
                       },

     

       268
       268
       +
                   ),

     

       269
       269
       +
                   Item::new(

     

       270
       270
       +
                       1015,

     

       271
       271
       +
                       &TestData {

     

       272
       272
       +
                           id: 3,

     

       273
       273
       +
                           value: "third".to_string(),

     

       274
       274
       +
                       },

     

       275
       275
       +
                   ),

     

       276
       276
       +
                   Item::new(

     

       277
       277
       +
                       1025,

     

       278
       278
       +
                       &TestData {

     

       279
       279
       +
                           id: 4,

     

       280
       280
       +
                           value: "fourth".to_string(),

     

       281
       281
       +
                       },

     

       282
       282
       +
                   ),

     

       283
       283
       +
               ];

     

       284
       284
       +
       

     

       285
       285
       +
               // encode

     

       286
       286
       +
               let mut buffer = Vec::new();

     

       287
       287
       +
               let mut encoder = ItemEncoder::new(&mut buffer);

     

       288
       288
       +
       

     

       289
       289
       +
               for item in &items {

     

       290
       290
       +
                   encoder.encode(item).unwrap();

     

       291
       291
       +
               }

     

       292
       292
       +
               encoder.finish().unwrap();

     

       293
       293
       +
       

     

       294
       294
       +
               // decode

     

       295
       295
       +
               let cursor = Cursor::new(buffer);

     

       296
       296
       +
               let mut decoder = ItemDecoder::<_, TestData>::new(cursor, 1000).unwrap();

     

       297
       297
       +
       

     

       298
       298
       +
               let mut decoded_items = Vec::new();

     

       299
       299
       +
               while let Some(item) = decoder.decode().unwrap() {

     

       300
       300
       +
                   decoded_items.push(item);

     

       301
       301
       +
               }

     

       302
       302
       +
       

     

       303
       303
       +
               assert_eq!(decoded_items.len(), 4);

     

       304
       304
       +
       

     

       305
       305
       +
               for (original, decoded) in items.iter().zip(decoded_items.iter()) {

     

       306
       306
       +
                   assert_eq!(original.timestamp, decoded.timestamp);

     

       307
       307
       +
                   assert_eq!(original.access().id, decoded.access().id);

     

       308
       308
       +
                   assert_eq!(

     

       309
       309
       +
                       original.access().value.as_str(),

     

       310
       310
       +
                       decoded.access().value.as_str()

     

       311
       311
       +
                   );

     

       312
       312
       +
               }

     

       313
       313
       +
           }

     

       314
       314
       +
       

     

       315
       315
       +
           #[test]

     

       316
       316
       +
           fn test_encoder_decoder_with_iterator() {

     

       317
       317
       +
               let items = vec![

     

       318
       318
       +
                   Item::new(

     

       319
       319
       +
                       2000,

     

       320
       320
       +
                       &TestData {

     

       321
       321
       +
                           id: 10,

     

       322
       322
       +
                           value: "a".to_string(),

     

       323
       323
       +
                       },

     

       324
       324
       +
                   ),

     

       325
       325
       +
                   Item::new(

     

       326
       326
       +
                       2005,

     

       327
       327
       +
                       &TestData {

     

       328
       328
       +
                           id: 20,

     

       329
       329
       +
                           value: "b".to_string(),

     

       330
       330
       +
                       },

     

       331
       331
       +
                   ),

     

       332
       332
       +
                   Item::new(

     

       333
       333
       +
                       2012,

     

       334
       334
       +
                       &TestData {

     

       335
       335
       +
                           id: 30,

     

       336
       336
       +
                           value: "c".to_string(),

     

       337
       337
       +
                       },

     

       338
       338
       +
                   ),

     

       339
       339
       +
               ];

     

       340
       340
       +
       

     

       341
       341
       +
               // encode

     

       342
       342
       +
               let mut buffer = Vec::new();

     

       343
       343
       +
               let mut encoder = ItemEncoder::new(&mut buffer);

     

       344
       344
       +
       

     

       345
       345
       +
               for item in &items {

     

       346
       346
       +
                   encoder.encode(item).unwrap();

     

       347
       347
       +
               }

     

       348
       348
       +
               encoder.finish().unwrap();

     

       349
       349
       +
       

     

       350
       350
       +
               // decode

     

       351
       351
       +
               let cursor = Cursor::new(buffer);

     

       352
       352
       +
               let decoder = ItemDecoder::<_, TestData>::new(cursor, 2000).unwrap();

     

       353
       353
       +
       

     

       354
       354
       +
               let decoded_items: Result<Vec<_>, _> = decoder.collect();

     

       355
       355
       +
               let decoded_items = decoded_items.unwrap();

     

       356
       356
       +
       

     

       357
       357
       +
               assert_eq!(decoded_items.len(), 3);

     

       358
       358
       +
               assert_eq!(decoded_items[0].timestamp, 2000);

     

       359
       359
       +
               assert_eq!(decoded_items[1].timestamp, 2005);

     

       360
       360
       +
               assert_eq!(decoded_items[2].timestamp, 2012);

     

       361
       361
       +
       

     

       362
       362
       +
               assert_eq!(decoded_items[0].access().id, 10);

     

       363
       363
       +
               assert_eq!(decoded_items[1].access().id, 20);

     

       364
       364
       +
               assert_eq!(decoded_items[2].access().id, 30);

     

       365
       365
       +
           }

     

       366
       366
       +
       

     

       367
       367
       +
           #[test]

     

       368
       368
       +
           fn test_delta_compression() {

     

       369
       369
       +
               let items = vec![

     

       370
       370
       +
                   Item::new(

     

       371
       371
       +
                       1000,

     

       372
       372
       +
                       &TestData {

     

       373
       373
       +
                           id: 1,

     

       374
       374
       +
                           value: "a".to_string(),

     

       375
       375
       +
                       },

     

       376
       376
       +
                   ),

     

       377
       377
       +
                   Item::new(

     

       378
       378
       +
                       1010,

     

       379
       379
       +
                       &TestData {

     

       380
       380
       +
                           id: 2,

     

       381
       381
       +
                           value: "b".to_string(),

     

       382
       382
       +
                       },

     

       383
       383
       +
                   ), // delta = 10

     

       384
       384
       +
                   Item::new(

     

       385
       385
       +
                       1020,

     

       386
       386
       +
                       &TestData {

     

       387
       387
       +
                           id: 3,

     

       388
       388
       +
                           value: "c".to_string(),

     

       389
       389
       +
                       },

     

       390
       390
       +
                   ), // delta = 10, delta-of-delta = 0

     

       391
       391
       +
                   Item::new(

     

       392
       392
       +
                       1025,

     

       393
       393
       +
                       &TestData {

     

       394
       394
       +
                           id: 4,

     

       395
       395
       +
                           value: "d".to_string(),

     

       396
       396
       +
                       },

     

       397
       397
       +
                   ), // delta = 5, delta-of-delta = -5

     

       398
       398
       +
               ];

     

       399
       399
       +
       

     

       400
       400
       +
               let mut buffer = Vec::new();

     

       401
       401
       +
               let mut encoder = ItemEncoder::new(&mut buffer);

     

       402
       402
       +
       

     

       403
       403
       +
               for item in &items {

     

       404
       404
       +
                   encoder.encode(item).unwrap();

     

       405
       405
       +
               }

     

       406
       406
       +
               encoder.finish().unwrap();

     

       407
       407
       +
       

     

       408
       408
       +
               // decode and verify

     

       409
       409
       +
               let cursor = Cursor::new(buffer);

     

       410
       410
       +
               let decoder = ItemDecoder::<_, TestData>::new(cursor, 1000).unwrap();

     

       411
       411
       +
       

     

       412
       412
       +
               let decoded_items: Result<Vec<_>, _> = decoder.collect();

     

       413
       413
       +
               let decoded_items = decoded_items.unwrap();

     

       414
       414
       +
       

     

       415
       415
       +
               for (original, decoded) in items.iter().zip(decoded_items.iter()) {

     

       416
       416
       +
                   assert_eq!(original.timestamp, decoded.timestamp);

     

       417
       417
       +
                   assert_eq!(original.access().id, decoded.access().id);

     

       418
       418
       +
               }

     

       419
       419
       +
           }

     

       420
       420
       +
       

     

       421
       421
       +
           #[test]

     

       422
       422
       +
           fn test_empty_decode() {

     

       423
       423
       +
               let buffer = Vec::new();

     

       424
       424
       +
               let cursor = Cursor::new(buffer);

     

       425
       425
       +
               let mut decoder = ItemDecoder::<_, TestData>::new(cursor, 1000).unwrap();

     

       426
       426
       +
       

     

       427
       427
       +
               let result = decoder.decode().unwrap();

     

       428
       428
       +
               assert!(result.is_none());

     

       429
       429
       +
           }

     

       430
       430
       +
       

     

       431
       431
       +
           #[test]

     

       432
       432
       +
           fn test_backwards_timestamp() {

     

       433
       433
       +
               let items = vec![

     

       434
       434
       +
                   Item::new(

     

       435
       435
       +
                       1000,

     

       436
       436
       +
                       &TestData {

     

       437
       437
       +
                           id: 1,

     

       438
       438
       +
                           value: "first".to_string(),

     

       439
       439
       +
                       },

     

       440
       440
       +
                   ),

     

       441
       441
       +
                   Item::new(

     

       442
       442
       +
                       900,

     

       443
       443
       +
                       &TestData {

     

       444
       444
       +
                           id: 2,

     

       445
       445
       +
                           value: "second".to_string(),

     

       446
       446
       +
                       },

     

       447
       447
       +
                   ),

     

       448
       448
       +
               ];

     

       449
       449
       +
       

     

       450
       450
       +
               let mut buffer = Vec::new();

     

       451
       451
       +
               let mut encoder = ItemEncoder::new(&mut buffer);

     

       452
       452
       +
       

     

       453
       453
       +
               for item in &items {

     

       454
       454
       +
                   encoder.encode(item).unwrap();

     

       455
       455
       +
               }

     

       456
       456
       +
               encoder.finish().unwrap();

     

       457
       457
       +
       

     

       458
       458
       +
               let cursor = Cursor::new(buffer);

     

       459
       459
       +
               let decoder = ItemDecoder::<_, TestData>::new(cursor, 1000).unwrap();

     

       460
       460
       +
       

     

       461
       461
       +
               let decoded_items: Result<Vec<_>, _> = decoder.collect();

     

       462
       462
       +
               let decoded_items = decoded_items.unwrap();

     

       463
       463
       +
       

     

       464
       464
       +
               assert_eq!(decoded_items.len(), 2);

     

       465
       465
       +
               assert_eq!(decoded_items[0].timestamp, 1000);

     

       466
       466
       +
               assert_eq!(decoded_items[1].timestamp, 900);

     

       467
       467
       +
           }

     

       468
       468
       +
       

     

       469
       469
       +
           #[test]

     

       470
       470
       +
           fn test_different_data_sizes() {

     

       471
       471
       +
               let small_data = TestData {

     

       472
       472
       +
                   id: 1,

     

       473
       473
       +
                   value: "x".to_string(),

     

       474
       474
       +
               };

     

       475
       475
       +
               let large_data = TestData {

     

       476
       476
       +
                   id: 2,

     

       477
       477
       +
                   value: "a".repeat(1000),

     

       478
       478
       +
               };

     

       479
       479
       +
       

     

       480
       480
       +
               let items = vec![Item::new(1000, &small_data), Item::new(1001, &large_data)];

     

       481
       481
       +
       

     

       482
       482
       +
               let mut buffer = Vec::new();

     

       483
       483
       +
               let mut encoder = ItemEncoder::new(&mut buffer);

     

       484
       484
       +
       

     

       485
       485
       +
               for item in &items {

     

       486
       486
       +
                   encoder.encode(item).unwrap();

     

       487
       487
       +
               }

     

       488
       488
       +
               encoder.finish().unwrap();

     

       489
       489
       +
       

     

       490
       490
       +
               let cursor = Cursor::new(buffer);

     

       491
       491
       +
               let decoder = ItemDecoder::<_, TestData>::new(cursor, 1000).unwrap();

     

       492
       492
       +
       

     

       493
       493
       +
               let decoded_items: Result<Vec<_>, _> = decoder.collect();

     

       494
       494
       +
               let decoded_items = decoded_items.unwrap();

     

       495
       495
       +
       

     

       496
       496
       +
               assert_eq!(decoded_items.len(), 2);

     

       497
       497
       +
               assert_eq!(decoded_items[0].access().value.as_str(), "x");

     

       498
       498
       +
               assert_eq!(decoded_items[1].access().value.len(), 1000);

     

       499
       499
       +
               assert_eq!(decoded_items[1].access().value.as_str(), "a".repeat(1000));

     

       500
       500
       +
           }

     

       501
       501
       +
       }

+424

server/src/db_old/mod.rs

···

       1
       1
       +
       use std::{

     

       2
       2
       +
           io::Cursor,

     

       3
       3
       +
           ops::{Bound, Deref, RangeBounds},

     

       4
       4
       +
           path::Path,

     

       5
       5
       +
           sync::{

     

       6
       6
       +
               Arc,

     

       7
       7
       +
               atomic::{AtomicU64, AtomicUsize, Ordering as AtomicOrdering},

     

       8
       8
       +
           },

     

       9
       9
       +
           time::{Duration, Instant},

     

       10
       10
       +
       };

     

       11
       11
       +
       

     

       12
       12
       +
       use fjall::{Config, Keyspace, Partition, PartitionCreateOptions, Slice};

     

       13
       13
       +
       use ordered_varint::Variable;

     

       14
       14
       +
       use rkyv::{Archive, Deserialize, Serialize, rancor::Error};

     

       15
       15
       +
       use smol_str::SmolStr;

     

       16
       16
       +
       use tokio::sync::broadcast;

     

       17
       17
       +
       

     

       18
       18
       +
       use crate::{

     

       19
       19
       +
           db_old::block::{ReadVariableExt, WriteVariableExt},

     

       20
       20
       +
           error::{AppError, AppResult},

     

       21
       21
       +
           jetstream::JetstreamEvent,

     

       22
       22
       +
           utils::{DefaultRateTracker, get_time},

     

       23
       23
       +
       };

     

       24
       24
       +
       

     

       25
       25
       +
       mod block;

     

       26
       26
       +
       

     

       27
       27
       +
       #[derive(Clone, Debug, Default, Archive, Deserialize, Serialize, PartialEq)]

     

       28
       28
       +
       #[rkyv(compare(PartialEq), derive(Debug))]

     

       29
       29
       +
       pub struct NsidCounts {

     

       30
       30
       +
           pub count: u128,

     

       31
       31
       +
           pub deleted_count: u128,

     

       32
       32
       +
           pub last_seen: u64,

     

       33
       33
       +
       }

     

       34
       34
       +
       

     

       35
       35
       +
       #[derive(Debug, Default, Archive, Deserialize, Serialize, PartialEq)]

     

       36
       36
       +
       #[rkyv(compare(PartialEq), derive(Debug))]

     

       37
       37
       +
       pub struct NsidHit {

     

       38
       38
       +
           pub deleted: bool,

     

       39
       39
       +
       }

     

       40
       40
       +
       

     

       41
       41
       +
       #[derive(Clone)]

     

       42
       42
       +
       pub struct EventRecord {

     

       43
       43
       +
           pub nsid: SmolStr,

     

       44
       44
       +
           pub timestamp: u64, // seconds

     

       45
       45
       +
           pub deleted: bool,

     

       46
       46
       +
       }

     

       47
       47
       +
       

     

       48
       48
       +
       impl EventRecord {

     

       49
       49
       +
           pub fn from_jetstream(event: JetstreamEvent) -> Option<Self> {

     

       50
       50
       +
               match event {

     

       51
       51
       +
                   JetstreamEvent::Commit {

     

       52
       52
       +
                       time_us, commit, ..

     

       53
       53
       +
                   } => Some(Self {

     

       54
       54
       +
                       nsid: commit.collection.into(),

     

       55
       55
       +
                       timestamp: time_us / 1_000_000,

     

       56
       56
       +
                       deleted: false,

     

       57
       57
       +
                   }),

     

       58
       58
       +
                   JetstreamEvent::Delete {

     

       59
       59
       +
                       time_us, commit, ..

     

       60
       60
       +
                   } => Some(Self {

     

       61
       61
       +
                       nsid: commit.collection.into(),

     

       62
       62
       +
                       timestamp: time_us / 1_000_000,

     

       63
       63
       +
                       deleted: true,

     

       64
       64
       +
                   }),

     

       65
       65
       +
                   _ => None,

     

       66
       66
       +
               }

     

       67
       67
       +
           }

     

       68
       68
       +
       }

     

       69
       69
       +
       

     

       70
       70
       +
       type ItemDecoder = block::ItemDecoder<Cursor<Slice>, NsidHit>;

     

       71
       71
       +
       type ItemEncoder = block::ItemEncoder<Vec<u8>, NsidHit>;

     

       72
       72
       +
       type Item = block::Item<NsidHit>;

     

       73
       73
       +
       

     

       74
       74
       +
       pub struct LexiconHandle {

     

       75
       75
       +
           tree: Partition,

     

       76
       76
       +
           buf: Arc<scc::Queue<EventRecord>>,

     

       77
       77
       +
           buf_len: AtomicUsize,

     

       78
       78
       +
           last_insert: AtomicU64,

     

       79
       79
       +
           eps: DefaultRateTracker,

     

       80
       80
       +
           block_size: AtomicUsize,

     

       81
       81
       +
       }

     

       82
       82
       +
       

     

       83
       83
       +
       impl LexiconHandle {

     

       84
       84
       +
           fn new(keyspace: &Keyspace, nsid: &str) -> Self {

     

       85
       85
       +
               let opts = PartitionCreateOptions::default().compression(fjall::CompressionType::Miniz(9));

     

       86
       86
       +
               Self {

     

       87
       87
       +
                   tree: keyspace.open_partition(nsid, opts).unwrap(),

     

       88
       88
       +
                   buf: Default::default(),

     

       89
       89
       +
                   buf_len: AtomicUsize::new(0),

     

       90
       90
       +
                   last_insert: AtomicU64::new(0),

     

       91
       91
       +
                   eps: DefaultRateTracker::new(Duration::from_secs(5)),

     

       92
       92
       +
                   block_size: AtomicUsize::new(1000),

     

       93
       93
       +
               }

     

       94
       94
       +
           }

     

       95
       95
       +
       

     

       96
       96
       +
           fn item_count(&self) -> usize {

     

       97
       97
       +
               self.buf_len.load(AtomicOrdering::Acquire)

     

       98
       98
       +
           }

     

       99
       99
       +
       

     

       100
       100
       +
           fn last_insert(&self) -> u64 {

     

       101
       101
       +
               self.last_insert.load(AtomicOrdering::Acquire)

     

       102
       102
       +
           }

     

       103
       103
       +
       

     

       104
       104
       +
           fn suggested_block_size(&self) -> usize {

     

       105
       105
       +
               self.block_size.load(AtomicOrdering::Relaxed)

     

       106
       106
       +
           }

     

       107
       107
       +
       

     

       108
       108
       +
           fn insert(&self, event: EventRecord) {

     

       109
       109
       +
               self.buf.push(event);

     

       110
       110
       +
               self.buf_len.fetch_add(1, AtomicOrdering::Release);

     

       111
       111
       +
               self.last_insert

     

       112
       112
       +
                   .store(get_time().as_millis() as u64, AtomicOrdering::Release);

     

       113
       113
       +
               self.eps.observe(1);

     

       114
       114
       +
               let rate = self.eps.rate() as usize;

     

       115
       115
       +
               if rate != 0 {

     

       116
       116
       +
                   self.block_size.store(rate * 60, AtomicOrdering::Relaxed);

     

       117
       117
       +
               }

     

       118
       118
       +
           }

     

       119
       119
       +
       

     

       120
       120
       +
           fn sync(&self, max_block_size: usize) -> AppResult<usize> {

     

       121
       121
       +
               let mut writer = ItemEncoder::new(Vec::with_capacity(

     

       122
       122
       +
                   size_of::<u64>() + self.item_count().min(max_block_size) * size_of::<(u64, NsidHit)>(),

     

       123
       123
       +
               ));

     

       124
       124
       +
               let mut start_timestamp = None;

     

       125
       125
       +
               let mut end_timestamp = None;

     

       126
       126
       +
               let mut written = 0_usize;

     

       127
       127
       +
               while let Some(event) = self.buf.pop() {

     

       128
       128
       +
                   let item = Item::new(

     

       129
       129
       +
                       event.timestamp,

     

       130
       130
       +
                       &NsidHit {

     

       131
       131
       +
                           deleted: event.deleted,

     

       132
       132
       +
                       },

     

       133
       133
       +
                   );

     

       134
       134
       +
                   writer.encode(&item)?;

     

       135
       135
       +
                   if start_timestamp.is_none() {

     

       136
       136
       +
                       start_timestamp = Some(event.timestamp);

     

       137
       137
       +
                   }

     

       138
       138
       +
                   end_timestamp = Some(event.timestamp);

     

       139
       139
       +
                   if written >= max_block_size {

     

       140
       140
       +
                       break;

     

       141
       141
       +
                   }

     

       142
       142
       +
                   written += 1;

     

       143
       143
       +
               }

     

       144
       144
       +
               if let (Some(start_timestamp), Some(end_timestamp)) = (start_timestamp, end_timestamp) {

     

       145
       145
       +
                   self.buf_len.store(0, AtomicOrdering::Release);

     

       146
       146
       +
                   let value = writer.finish()?;

     

       147
       147
       +
                   let mut key = Vec::with_capacity(size_of::<u64>() * 2);

     

       148
       148
       +
                   key.write_varint(start_timestamp)?;

     

       149
       149
       +
                   key.write_varint(end_timestamp)?;

     

       150
       150
       +
                   self.tree.insert(key, value)?;

     

       151
       151
       +
               }

     

       152
       152
       +
               Ok(written)

     

       153
       153
       +
           }

     

       154
       154
       +
       }

     

       155
       155
       +
       

     

       156
       156
       +
       type BoxedIter<T> = Box<dyn Iterator<Item = T>>;

     

       157
       157
       +
       

     

       158
       158
       +
       // counts is nsid -> NsidCounts

     

       159
       159
       +
       // hits is tree per nsid: varint start time + varint end time -> block of hits

     

       160
       160
       +
       pub struct Db {

     

       161
       161
       +
           inner: Keyspace,

     

       162
       162
       +
           hits: scc::HashIndex<SmolStr, Arc<LexiconHandle>>,

     

       163
       163
       +
           counts: Partition,

     

       164
       164
       +
           event_broadcaster: broadcast::Sender<(SmolStr, NsidCounts)>,

     

       165
       165
       +
           eps: DefaultRateTracker,

     

       166
       166
       +
           min_block_size: usize,

     

       167
       167
       +
           max_block_size: usize,

     

       168
       168
       +
           max_last_activity: Duration,

     

       169
       169
       +
       }

     

       170
       170
       +
       

     

       171
       171
       +
       impl Db {

     

       172
       172
       +
           pub fn new(path: impl AsRef<Path>) -> AppResult<Self> {

     

       173
       173
       +
               tracing::info!("opening db...");

     

       174
       174
       +
               let ks = Config::new(path)

     

       175
       175
       +
                   .cache_size(8 * 1024 * 1024) // from talna

     

       176
       176
       +
                   .open()?;

     

       177
       177
       +
               Ok(Self {

     

       178
       178
       +
                   hits: Default::default(),

     

       179
       179
       +
                   counts: ks.open_partition(

     

       180
       180
       +
                       "_counts",

     

       181
       181
       +
                       PartitionCreateOptions::default().compression(fjall::CompressionType::None),

     

       182
       182
       +
                   )?,

     

       183
       183
       +
                   inner: ks,

     

       184
       184
       +
                   event_broadcaster: broadcast::channel(1000).0,

     

       185
       185
       +
                   eps: DefaultRateTracker::new(Duration::from_secs(1)),

     

       186
       186
       +
                   min_block_size: 512,

     

       187
       187
       +
                   max_block_size: 100_000,

     

       188
       188
       +
                   max_last_activity: Duration::from_secs(10),

     

       189
       189
       +
               })

     

       190
       190
       +
           }

     

       191
       191
       +
       

     

       192
       192
       +
           pub fn sync(&self, all: bool) -> AppResult<()> {

     

       193
       193
       +
               let _guard = scc::ebr::Guard::new();

     

       194
       194
       +
               for (nsid, tree) in self.hits.iter(&_guard) {

     

       195
       195
       +
                   let count = tree.item_count();

     

       196
       196
       +
                   let is_max_block_size = count > self.min_block_size.max(tree.suggested_block_size());

     

       197
       197
       +
                   let is_too_old = (get_time().as_millis() as u64 - tree.last_insert())

     

       198
       198
       +
                       > self.max_last_activity.as_millis() as u64;

     

       199
       199
       +
                   if count > 0 && (all || is_max_block_size || is_too_old) {

     

       200
       200
       +
                       loop {

     

       201
       201
       +
                           let synced = tree.sync(self.max_block_size)?;

     

       202
       202
       +
                           if synced == 0 {

     

       203
       203
       +
                               break;

     

       204
       204
       +
                           }

     

       205
       205
       +
                           tracing::info!("synced {synced} of {nsid} to db");

     

       206
       206
       +
                       }

     

       207
       207
       +
                   }

     

       208
       208
       +
               }

     

       209
       209
       +
               Ok(())

     

       210
       210
       +
           }

     

       211
       211
       +
       

     

       212
       212
       +
           #[inline(always)]

     

       213
       213
       +
           pub fn eps(&self) -> usize {

     

       214
       214
       +
               self.eps.rate() as usize

     

       215
       215
       +
           }

     

       216
       216
       +
       

     

       217
       217
       +
           #[inline(always)]

     

       218
       218
       +
           pub fn new_listener(&self) -> broadcast::Receiver<(SmolStr, NsidCounts)> {

     

       219
       219
       +
               self.event_broadcaster.subscribe()

     

       220
       220
       +
           }

     

       221
       221
       +
       

     

       222
       222
       +
           #[inline(always)]

     

       223
       223
       +
           fn maybe_run_in_nsid_tree<T>(

     

       224
       224
       +
               &self,

     

       225
       225
       +
               nsid: &str,

     

       226
       226
       +
               f: impl FnOnce(&LexiconHandle) -> T,

     

       227
       227
       +
           ) -> Option<T> {

     

       228
       228
       +
               let _guard = scc::ebr::Guard::new();

     

       229
       229
       +
               let handle = match self.hits.peek(nsid, &_guard) {

     

       230
       230
       +
                   Some(handle) => handle.clone(),

     

       231
       231
       +
                   None => {

     

       232
       232
       +
                       if self.inner.partition_exists(nsid) {

     

       233
       233
       +
                           let handle = Arc::new(LexiconHandle::new(&self.inner, nsid));

     

       234
       234
       +
                           let _ = self.hits.insert(SmolStr::new(nsid), handle.clone());

     

       235
       235
       +
                           handle

     

       236
       236
       +
                       } else {

     

       237
       237
       +
                           return None;

     

       238
       238
       +
                       }

     

       239
       239
       +
                   }

     

       240
       240
       +
               };

     

       241
       241
       +
               Some(f(&handle))

     

       242
       242
       +
           }

     

       243
       243
       +
       

     

       244
       244
       +
           #[inline(always)]

     

       245
       245
       +
           fn run_in_nsid_tree<T>(

     

       246
       246
       +
               &self,

     

       247
       247
       +
               nsid: SmolStr,

     

       248
       248
       +
               f: impl FnOnce(&LexiconHandle) -> AppResult<T>,

     

       249
       249
       +
           ) -> AppResult<T> {

     

       250
       250
       +
               f(self

     

       251
       251
       +
                   .hits

     

       252
       252
       +
                   .entry(nsid.clone())

     

       253
       253
       +
                   .or_insert_with(move || Arc::new(LexiconHandle::new(&self.inner, &nsid)))

     

       254
       254
       +
                   .get())

     

       255
       255
       +
           }

     

       256
       256
       +
       

     

       257
       257
       +
           pub fn record_event(&self, e: EventRecord) -> AppResult<()> {

     

       258
       258
       +
               let EventRecord {

     

       259
       259
       +
                   nsid,

     

       260
       260
       +
                   timestamp,

     

       261
       261
       +
                   deleted,

     

       262
       262
       +
               } = e.clone();

     

       263
       263
       +
       

     

       264
       264
       +
               // insert event

     

       265
       265
       +
               self.run_in_nsid_tree(nsid.clone(), move |tree| Ok(tree.insert(e)))?;

     

       266
       266
       +
               // increment count

     

       267
       267
       +
               let mut counts = self.get_count(&nsid)?;

     

       268
       268
       +
               counts.last_seen = timestamp;

     

       269
       269
       +
               if deleted {

     

       270
       270
       +
                   counts.deleted_count += 1;

     

       271
       271
       +
               } else {

     

       272
       272
       +
                   counts.count += 1;

     

       273
       273
       +
               }

     

       274
       274
       +
               self.insert_count(&nsid, counts.clone())?;

     

       275
       275
       +
               if self.event_broadcaster.receiver_count() > 0 {

     

       276
       276
       +
                   let _ = self.event_broadcaster.send((SmolStr::new(&nsid), counts));

     

       277
       277
       +
               }

     

       278
       278
       +
               self.eps.observe(1);

     

       279
       279
       +
               Ok(())

     

       280
       280
       +
           }

     

       281
       281
       +
       

     

       282
       282
       +
           #[inline(always)]

     

       283
       283
       +
           fn insert_count(&self, nsid: &str, counts: NsidCounts) -> AppResult<()> {

     

       284
       284
       +
               self.counts

     

       285
       285
       +
                   .insert(

     

       286
       286
       +
                       nsid,

     

       287
       287
       +
                       unsafe { rkyv::to_bytes::<Error>(&counts).unwrap_unchecked() }.as_slice(),

     

       288
       288
       +
                   )

     

       289
       289
       +
                   .map_err(AppError::from)

     

       290
       290
       +
           }

     

       291
       291
       +
       

     

       292
       292
       +
           pub fn get_count(&self, nsid: &str) -> AppResult<NsidCounts> {

     

       293
       293
       +
               let Some(raw) = self.counts.get(nsid)? else {

     

       294
       294
       +
                   return Ok(NsidCounts::default());

     

       295
       295
       +
               };

     

       296
       296
       +
               Ok(unsafe { rkyv::from_bytes_unchecked::<_, Error>(&raw).unwrap_unchecked() })

     

       297
       297
       +
           }

     

       298
       298
       +
       

     

       299
       299
       +
           pub fn get_counts(&self) -> impl Iterator<Item = AppResult<(SmolStr, NsidCounts)>> {

     

       300
       300
       +
               self.counts.iter().map(|res| {

     

       301
       301
       +
                   res.map_err(AppError::from).map(|(key, val)| {

     

       302
       302
       +
                       (

     

       303
       303
       +
                           SmolStr::new(unsafe { str::from_utf8_unchecked(&key) }),

     

       304
       304
       +
                           unsafe { rkyv::from_bytes_unchecked::<_, Error>(&val).unwrap_unchecked() },

     

       305
       305
       +
                       )

     

       306
       306
       +
                   })

     

       307
       307
       +
               })

     

       308
       308
       +
           }

     

       309
       309
       +
       

     

       310
       310
       +
           pub fn get_nsids(&self) -> impl Iterator<Item = impl Deref<Target = str> + 'static> {

     

       311
       311
       +
               self.inner

     

       312
       312
       +
                   .list_partitions()

     

       313
       313
       +
                   .into_iter()

     

       314
       314
       +
                   .filter(|k| k.deref() != "_counts")

     

       315
       315
       +
           }

     

       316
       316
       +
       

     

       317
       317
       +
           pub fn get_hits_debug(&self, nsid: &str) -> BoxedIter<AppResult<(Slice, Slice)>> {

     

       318
       318
       +
               self.maybe_run_in_nsid_tree(nsid, |handle| -> BoxedIter<AppResult<(Slice, Slice)>> {

     

       319
       319
       +
                   Box::new(

     

       320
       320
       +
                       handle

     

       321
       321
       +
                           .tree

     

       322
       322
       +
                           .iter()

     

       323
       323
       +
                           .rev()

     

       324
       324
       +
                           .map(|res| res.map_err(AppError::from)),

     

       325
       325
       +
                   )

     

       326
       326
       +
               })

     

       327
       327
       +
               .unwrap_or_else(|| Box::new(std::iter::empty()))

     

       328
       328
       +
           }

     

       329
       329
       +
       

     

       330
       330
       +
           pub fn get_hits(

     

       331
       331
       +
               &self,

     

       332
       332
       +
               nsid: &str,

     

       333
       333
       +
               range: impl RangeBounds<u64> + std::fmt::Debug,

     

       334
       334
       +
           ) -> BoxedIter<AppResult<Item>> {

     

       335
       335
       +
               let start = range

     

       336
       336
       +
                   .start_bound()

     

       337
       337
       +
                   .cloned()

     

       338
       338
       +
                   .map(|t| unsafe { t.to_variable_vec().unwrap_unchecked() });

     

       339
       339
       +
               let end = range

     

       340
       340
       +
                   .end_bound()

     

       341
       341
       +
                   .cloned()

     

       342
       342
       +
                   .map(|t| unsafe { t.to_variable_vec().unwrap_unchecked() });

     

       343
       343
       +
               let limit = match range.end_bound().cloned() {

     

       344
       344
       +
                   Bound::Included(end) => end,

     

       345
       345
       +
                   Bound::Excluded(end) => end.saturating_sub(1),

     

       346
       346
       +
                   Bound::Unbounded => u64::MAX,

     

       347
       347
       +
               };

     

       348
       348
       +
       

     

       349
       349
       +
               self.maybe_run_in_nsid_tree(nsid, move |handle| -> BoxedIter<AppResult<Item>> {

     

       350
       350
       +
                   let map_block = move |(key, val)| {

     

       351
       351
       +
                       let mut key_reader = Cursor::new(key);

     

       352
       352
       +
                       let start_timestamp = key_reader.read_varint::<u64>()?;

     

       353
       353
       +
                       let items =

     

       354
       354
       +
                           ItemDecoder::new(Cursor::new(val), start_timestamp)?.take_while(move |item| {

     

       355
       355
       +
                               item.as_ref().map_or(true, |item| item.timestamp <= limit)

     

       356
       356
       +
                           });

     

       357
       357
       +
                       Ok(items)

     

       358
       358
       +
                   };

     

       359
       359
       +
       

     

       360
       360
       +
                   Box::new(

     

       361
       361
       +
                       handle

     

       362
       362
       +
                           .tree

     

       363
       363
       +
                           .range(TimestampRange { start, end })

     

       364
       364
       +
                           .map(move |res| res.map_err(AppError::from).and_then(map_block))

     

       365
       365
       +
                           .flatten()

     

       366
       366
       +
                           .flatten(),

     

       367
       367
       +
                   )

     

       368
       368
       +
               })

     

       369
       369
       +
               .unwrap_or_else(|| Box::new(std::iter::empty()))

     

       370
       370
       +
           }

     

       371
       371
       +
       

     

       372
       372
       +
           pub fn tracking_since(&self) -> AppResult<u64> {

     

       373
       373
       +
               // HACK: we should actually store when we started tracking but im lazy

     

       374
       374
       +
               // should be accurate enough

     

       375
       375
       +
               self.maybe_run_in_nsid_tree("app.bsky.feed.like", |handle| {

     

       376
       376
       +
                   let Some((timestamps_raw, _)) = handle.tree.first_key_value()? else {

     

       377
       377
       +
                       return Ok(0);

     

       378
       378
       +
                   };

     

       379
       379
       +
                   let mut timestamp_reader = Cursor::new(timestamps_raw);

     

       380
       380
       +
                   timestamp_reader

     

       381
       381
       +
                       .read_varint::<u64>()

     

       382
       382
       +
                       .map_err(AppError::from)

     

       383
       383
       +
               })

     

       384
       384
       +
               .unwrap_or(Ok(0))

     

       385
       385
       +
           }

     

       386
       386
       +
       }

     

       387
       387
       +
       

     

       388
       388
       +
       type TimestampRepr = Vec<u8>;

     

       389
       389
       +
       

     

       390
       390
       +
       struct TimestampRange {

     

       391
       391
       +
           start: Bound<TimestampRepr>,

     

       392
       392
       +
           end: Bound<TimestampRepr>,

     

       393
       393
       +
       }

     

       394
       394
       +
       

     

       395
       395
       +
       impl RangeBounds<TimestampRepr> for TimestampRange {

     

       396
       396
       +
           #[inline(always)]

     

       397
       397
       +
           fn start_bound(&self) -> Bound<&TimestampRepr> {

     

       398
       398
       +
               self.start.as_ref()

     

       399
       399
       +
           }

     

       400
       400
       +
       

     

       401
       401
       +
           #[inline(always)]

     

       402
       402
       +
           fn end_bound(&self) -> Bound<&TimestampRepr> {

     

       403
       403
       +
               self.end.as_ref()

     

       404
       404
       +
           }

     

       405
       405
       +
       }

     

       406
       406
       +
       

     

       407
       407
       +
       type TimestampReprOld = [u8; 8];

     

       408
       408
       +
       

     

       409
       409
       +
       struct TimestampRangeOld {

     

       410
       410
       +
           start: Bound<TimestampReprOld>,

     

       411
       411
       +
           end: Bound<TimestampReprOld>,

     

       412
       412
       +
       }

     

       413
       413
       +
       

     

       414
       414
       +
       impl RangeBounds<TimestampReprOld> for TimestampRangeOld {

     

       415
       415
       +
           #[inline(always)]

     

       416
       416
       +
           fn start_bound(&self) -> Bound<&TimestampReprOld> {

     

       417
       417
       +
               self.start.as_ref()

     

       418
       418
       +
           }

     

       419
       419
       +
       

     

       420
       420
       +
           #[inline(always)]

     

       421
       421
       +
           fn end_bound(&self) -> Bound<&TimestampReprOld> {

     

       422
       422
       +
               self.end.as_ref()

     

       423
       423
       +
           }

     

       424
       424
       +
       }

+13 -35

server/src/main.rs

···

       1
       1
       -
       use std::{ops::Deref, time::Duration, u64, usize};

     

       1
       1
       +
       use std::{ops::Deref, time::Duration, u64};

     

       2
       2
        
       

     

       3
       3
        
       use itertools::Itertools;

     

       4
       4
        
       use rclite::Arc;

     
···

       17
       17
        
       

     

       18
       18
        
       mod api;

     

       19
       19
        
       mod db;

     

       20
       20
       +
       mod db_old;

     

       20
       21
        
       mod error;

     

       21
       22
        
       mod jetstream;

     

       22
       23
        
       mod utils;

     
···

       25
       26
        
       #[global_allocator]

     

       26
       27
        
       static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;

     

       27
       28
        
       

     

       29
       29
       +
       #[cfg(target_env = "msvc")]

     

       30
       30
       +
       #[global_allocator]

     

       31
       31
       +
       static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;

     

       32
       32
       +
       

     

       28
       33
        
       #[tokio::main]

     

       29
       34
        
       async fn main() {

     

       30
       35
        
           tracing_subscriber::fmt::fmt()

     
···

       47
       52
        
               }

     

       48
       53
        
               Some("debug") => {

     

       49
       54
        
                   debug();

     

       50
       50
       -
                   return;

     

       51
       51
       -
               }

     

       52
       52
       -
               Some("print") => {

     

       53
       53
       -
                   print_all();

     

       54
       55
        
                   return;

     

       55
       56
        
               }

     

       56
       57
        
               Some(x) => {

     
···

       106
       107
        
               move || {

     

       107
       108
        
                   let mut buffer = Vec::new();

     

       108
       109
        
                   loop {

     

       109
       109
       -
                       let read = event_rx.blocking_recv_many(&mut buffer, 500);

     

       110
       110
       +
                       let read = event_rx.blocking_recv_many(&mut buffer, 100);

     

       110
       111
        
                       if let Err(err) = db.ingest_events(buffer.drain(..)) {

     

       111
       112
        
                           tracing::error!("failed to ingest events: {}", err);

     

       112
       113
        
                       }

     
···

       152
       153
        
                                   if db.is_shutting_down() {

     

       153
       154
        
                                       return;

     

       154
       155
        
                                   }

     

       155
       155
       -
                                   let end = get_time();

     

       156
       156
       +
                                   let end = get_time() - compact_period / 2;

     

       156
       157
        
                                   let start = end - compact_period;

     

       157
       158
        
                                   let range = start.as_secs()..end.as_secs();

     

       158
       159
        
                                   tracing::info!(

     
···

       207
       208
        
           db.sync(true).expect("cant sync db");

     

       208
       209
        
       }

     

       209
       210
        
       

     

       210
       210
       -
       fn print_all() {

     

       211
       211
       -
           let db = Db::new(DbConfig::default(), CancellationToken::new()).expect("couldnt create db");

     

       212
       212
       -
           let nsids = db.get_nsids().collect::<Vec<_>>();

     

       213
       213
       -
           let mut count = 0_usize;

     

       214
       214
       -
           for nsid in nsids {

     

       215
       215
       -
               println!("{}:", nsid.deref());

     

       216
       216
       -
               for hit in db.get_hits(&nsid, .., usize::MAX) {

     

       217
       217
       -
                   let hit = hit.expect("aaa");

     

       218
       218
       -
                   println!("{} {}", hit.timestamp, hit.deser().unwrap().deleted);

     

       219
       219
       -
                   count += 1;

     

       220
       220
       -
               }

     

       221
       221
       -
           }

     

       222
       222
       -
           println!("total hits: {}", count);

     

       223
       223
       -
       }

     

       224
       224
       -
       

     

       225
       211
        
       fn debug() {

     

       226
       212
        
           let db = Db::new(DbConfig::default(), CancellationToken::new()).expect("couldnt create db");

     

       227
       213
        
           let info = db.info().expect("cant get db info");

     
···

       274
       260
        
       

     

       275
       261
        
       fn migrate() {

     

       276
       262
        
           let cancel_token = CancellationToken::new();

     

       277
       277
       -
           let from = Arc::new(

     

       278
       278
       -
               Db::new(

     

       279
       279
       -
                   DbConfig::default().path(".fjall_data_from"),

     

       280
       280
       -
                   cancel_token.child_token(),

     

       281
       281
       -
               )

     

       282
       282
       -
               .expect("couldnt create db"),

     

       283
       283
       -
           );

     

       263
       263
       +
       

     

       264
       264
       +
           let from = Arc::new(db_old::Db::new(".fjall_data_from").expect("couldnt create db"));

     

       265
       265
       +
       

     

       284
       266
        
           let to = Arc::new(

     

       285
       267
        
               Db::new(

     

       286
       268
        
                   DbConfig::default().path(".fjall_data_to").ks(|c| {

     
···

       295
       277
        
           );

     

       296
       278
        
       

     

       297
       279
        
           let nsids = from.get_nsids().collect::<Vec<_>>();

     

       298
       298
       -
           let _eps_thread = std::thread::spawn({

     

       280
       280
       +
           let eps_thread = std::thread::spawn({

     

       299
       281
        
               let to = to.clone();

     

       300
       282
        
               move || {

     

       301
       283
        
                   loop {

     
···

       315
       297
        
               threads.push(std::thread::spawn(move || {

     

       316
       298
        
                   tracing::info!("{}: migrating...", nsid.deref());

     

       317
       299
        
                   let mut count = 0_u64;

     

       318
       318
       -
                   for hits in from

     

       319
       319
       -
                       .get_hits(&nsid, .., usize::MAX)

     

       320
       320
       -
                       .chunks(100000)

     

       321
       321
       -
                       .into_iter()

     

       322
       322
       -
                   {

     

       300
       300
       +
                   for hits in from.get_hits(&nsid, ..).chunks(100000).into_iter() {

     

       323
       301
        
                       to.ingest_events(hits.map(|hit| {

     

       324
       302
        
                           count += 1;

     

       325
       303
        
                           let hit = hit.expect("cant decode hit");

-66

server/src/utils.rs

···

       1
       1
        
       use std::io::{self, Read, Write};

     

       2
       2
       -
       use std::ops::Deref;

     

       3
       2
        
       use std::sync::atomic::{AtomicU64, Ordering};

     

       4
       3
        
       use std::time::Duration;

     

       5
       4
        
       

     

       6
       6
       -
       use arc_swap::RefCnt;

     

       7
       5
        
       use byteview::ByteView;

     

       8
       6
        
       use ordered_varint::Variable;

     

       9
       9
       -
       use rclite::Arc;

     

       10
       7
        
       

     

       11
       8
        
       pub fn get_time() -> Duration {

     

       12
       9
        
           std::time::SystemTime::now()

     
···

       323
       320
        
               }

     

       324
       321
        
           }

     

       325
       322
        
       }

     

       326
       326
       -
       

     

       327
       327
       -
       pub type ArcliteSwap<T> = arc_swap::ArcSwapAny<ArcRefCnt<T>>;

     

       328
       328
       -
       

     

       329
       329
       -
       pub struct ArcRefCnt<T>(Arc<T>);

     

       330
       330
       -
       

     

       331
       331
       -
       impl<T> ArcRefCnt<T> {

     

       332
       332
       -
           pub fn new(value: T) -> Self {

     

       333
       333
       -
               Self(Arc::new(value))

     

       334
       334
       -
           }

     

       335
       335
       -
       }

     

       336
       336
       -
       

     

       337
       337
       -
       impl<T> Deref for ArcRefCnt<T> {

     

       338
       338
       -
           type Target = T;

     

       339
       339
       -
       

     

       340
       340
       -
           fn deref(&self) -> &Self::Target {

     

       341
       341
       -
               &self.0

     

       342
       342
       -
           }

     

       343
       343
       -
       }

     

       344
       344
       -
       

     

       345
       345
       -
       impl<T> Clone for ArcRefCnt<T> {

     

       346
       346
       -
           fn clone(&self) -> Self {

     

       347
       347
       -
               Self(self.0.clone())

     

       348
       348
       -
           }

     

       349
       349
       -
       }

     

       350
       350
       -
       

     

       351
       351
       -
       // SAFETY: uhhhhhhhh copied the Arc impl from arc_swap xd

     

       352
       352
       -
       unsafe impl<T> RefCnt for ArcRefCnt<T> {

     

       353
       353
       -
           type Base = T;

     

       354
       354
       -
       

     

       355
       355
       -
           fn into_ptr(me: Self) -> *mut Self::Base {

     

       356
       356
       -
               Arc::into_raw(me.0) as *mut T

     

       357
       357
       -
           }

     

       358
       358
       -
       

     

       359
       359
       -
           fn as_ptr(me: &Self) -> *mut Self::Base {

     

       360
       360
       -
               // Slightly convoluted way to do this, but this avoids stacked borrows violations. The same

     

       361
       361
       -
               // intention as

     

       362
       362
       -
               //

     

       363
       363
       -
               // me as &T as *const T as *mut T

     

       364
       364
       -
               //

     

       365
       365
       -
               // We first create a "shallow copy" of me - one that doesn't really own its ref count

     

       366
       366
       -
               // (that's OK, me _does_ own it, so it can't be destroyed in the meantime).

     

       367
       367
       -
               // Then we can use into_raw (which preserves not having the ref count).

     

       368
       368
       -
               //

     

       369
       369
       -
               // We need to "revert" the changes we did. In current std implementation, the combination

     

       370
       370
       -
               // of from_raw and forget is no-op. But formally, into_raw shall be paired with from_raw

     

       371
       371
       -
               // and that read shall be paired with forget to properly "close the brackets". In future

     

       372
       372
       -
               // versions of STD, these may become something else that's not really no-op (unlikely, but

     

       373
       373
       -
               // possible), so we future-proof it a bit.

     

       374
       374
       -
       

     

       375
       375
       -
               // SAFETY: &T cast to *const T will always be aligned, initialised and valid for reads

     

       376
       376
       -
               let ptr = Arc::into_raw(unsafe { std::ptr::read(&me.0) });

     

       377
       377
       -
               let ptr = ptr as *mut T;

     

       378
       378
       -
       

     

       379
       379
       -
               // SAFETY: We got the pointer from into_raw just above

     

       380
       380
       -
               std::mem::forget(unsafe { Arc::from_raw(ptr) });

     

       381
       381
       -
       

     

       382
       382
       -
               ptr

     

       383
       383
       -
           }

     

       384
       384
       -
       

     

       385
       385
       -
           unsafe fn from_ptr(ptr: *const Self::Base) -> Self {

     

       386
       386
       -
               Self(unsafe { Arc::from_raw(ptr) })

     

       387
       387
       -
           }

     

       388
       388
       -
       }

Compare changes