Drop the composite_index in favor of a multi-column unique index

Also set all the channel_update fields to NOT NULL, because they
should be non-null always.

diff --git a/src/config.rs b/src/config.rs
index 5369e8d04c89bf15eda6e1a4f76eb25a319c3521..b9f26b32cedaadd9cad301df1e45f4c06c690140 100644 (file)
--- a/src/config.rs
+++ b/src/config.rs
@@ -11,7 +11,7 @@ use crate::hex_utils;
 
 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;
 
@@ -60,21 +60,19 @@ pub(crate) fn db_announcement_table_creation_query() -> &'static str {
 }
 
 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()
        )"
 }
@@ -87,6 +85,7 @@ pub(crate) fn db_index_creation_query() -> &'static str {
        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);
        "
 }
 
@@ -175,6 +174,22 @@ pub(crate) async fn upgrade_db(schema: i32, client: &mut tokio_postgres::Client)
                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);
        }

diff --git a/src/hex_utils.rs b/src/hex_utils.rs
index 9d59aa91cece52a6a50c9d514fead7e6f1858685..36f2779b02f0efc7f04ba5077648c6301c9861bb 100644 (file)
--- a/src/hex_utils.rs
+++ b/src/hex_utils.rs
@@ -21,16 +21,6 @@ pub fn to_vec(hex: &str) -> Option<Vec<u8>> {
     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,

diff --git a/src/persistence.rs b/src/persistence.rs
index 404792d201504ef3140a9449d4d45c575ada2c96..51c242b7c8e94fab862fc305e8558e23cde719f7 100644 (file)
--- a/src/persistence.rs
+++ b/src/persistence.rs
@@ -7,7 +7,7 @@ use lightning::util::ser::Writeable;
 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 {
@@ -135,8 +135,6 @@ impl 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;
@@ -150,7 +148,6 @@ impl GossipPersister {
 
                                        let result = client
                                                .execute("INSERT INTO channel_updates (\
-                                                       composite_index, \
                                                        short_channel_id, \
                                                        timestamp, \
                                                        channel_flags, \
@@ -162,8 +159,7 @@ impl GossipPersister {
                                                        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,
                                                        &timestamp,
                                                        &(update.contents.flags as i16),