use futures::stream::{FuturesUnordered, StreamExt};
-pub(crate) const SCHEMA_VERSION: i32 = 6;
+pub(crate) const SCHEMA_VERSION: i32 = 7;
pub(crate) const SNAPSHOT_CALCULATION_INTERVAL: u32 = 3600 * 24; // every 24 hours, in seconds
pub(crate) const DOWNLOAD_NEW_GOSSIP: bool = true;
}
pub(crate) fn db_channel_update_table_creation_query() -> &'static str {
- // We'll run out of room in composite index at block 8,388,608 or in the year 2286
"CREATE TABLE IF NOT EXISTS channel_updates (
id SERIAL PRIMARY KEY,
- composite_index character(29) UNIQUE,
short_channel_id bigint NOT NULL,
- timestamp bigint,
- channel_flags smallint,
+ timestamp bigint NOT NULL,
+ channel_flags smallint NOT NULL,
direction boolean NOT NULL,
- disable boolean,
- cltv_expiry_delta integer,
- htlc_minimum_msat bigint,
- fee_base_msat integer,
- fee_proportional_millionths integer,
- htlc_maximum_msat bigint,
- blob_signed BYTEA,
+ disable boolean NOT NULL,
+ cltv_expiry_delta integer NOT NULL,
+ htlc_minimum_msat bigint NOT NULL,
+ fee_base_msat integer NOT NULL,
+ fee_proportional_millionths integer NOT NULL,
+ htlc_maximum_msat bigint NOT NULL,
+ blob_signed BYTEA NOT NULL,
seen timestamp NOT NULL DEFAULT NOW()
)"
}
CREATE INDEX IF NOT EXISTS channel_updates_seen ON channel_updates(seen);
CREATE INDEX IF NOT EXISTS channel_updates_scid_seen ON channel_updates(short_channel_id, seen);
CREATE INDEX IF NOT EXISTS channel_updates_scid_dir_seen ON channel_updates(short_channel_id ASC, direction ASC, seen DESC);
+ CREATE UNIQUE INDEX IF NOT EXISTS channel_updates_key ON channel_updates (short_channel_id, direction, timestamp);
"
}
tx.execute("UPDATE config SET db_schema = 6 WHERE id = 1", &[]).await.unwrap();
tx.commit().await.unwrap();
}
+ if schema >= 1 && schema <= 6 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("ALTER TABLE channel_updates DROP COLUMN composite_index", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER timestamp SET NOT NULL", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER channel_flags SET NOT NULL", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER disable SET NOT NULL", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER cltv_expiry_delta SET NOT NULL", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER htlc_minimum_msat SET NOT NULL", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER fee_base_msat SET NOT NULL", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER fee_proportional_millionths SET NOT NULL", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER htlc_maximum_msat SET NOT NULL", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER blob_signed SET NOT NULL", &[]).await.unwrap();
+ tx.execute("CREATE UNIQUE INDEX channel_updates_key ON channel_updates (short_channel_id, direction, timestamp)", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 7 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
if schema <= 1 || schema > SCHEMA_VERSION {
panic!("Unknown schema in db: {}, we support up to {}", schema, SCHEMA_VERSION);
}
Some(out)
}
-pub fn to_composite_index(scid: i64, timestamp: i64, direction: bool) -> String {
- let scid_be = scid.to_be_bytes();
- let res = format!("{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}:{}:{}",
- scid_be[0], scid_be[1], scid_be[2], scid_be[3],
- scid_be[4], scid_be[5], scid_be[6], scid_be[7],
- timestamp, direction as u8);
- assert_eq!(res.len(), 29); // Our SQL Type requires len of 29
- res
-}
-
pub fn to_compressed_pubkey(hex: &str) -> Option<PublicKey> {
let data = match to_vec(&hex[0..33 * 2]) {
Some(bytes) => bytes,
use tokio::sync::mpsc;
use tokio_postgres::NoTls;
-use crate::{config, hex_utils, TestLogger};
+use crate::{config, TestLogger};
use crate::types::GossipMessage;
pub(crate) struct GossipPersister {
let direction = (update.contents.flags & 1) == 1;
let disable = (update.contents.flags & 2) > 0;
- let composite_index = hex_utils::to_composite_index(scid, timestamp, direction);
-
let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
let fee_base_msat = update.contents.fee_base_msat as i32;
let result = client
.execute("INSERT INTO channel_updates (\
- composite_index, \
short_channel_id, \
timestamp, \
channel_flags, \
fee_proportional_millionths, \
htlc_maximum_msat, \
blob_signed \
- ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT (composite_index) DO NOTHING", &[
- &composite_index,
+ ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING", &[
&scid,
×tamp,
&(update.contents.flags as i16),