+pub(crate) async fn upgrade_db(schema: i32, client: &mut tokio_postgres::Client) {
+ if schema == 1 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("ALTER TABLE channel_updates DROP COLUMN chain_hash", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_announcements DROP COLUMN chain_hash", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 2 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
+ if schema == 1 || schema == 2 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("ALTER TABLE channel_updates DROP COLUMN short_channel_id", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ADD COLUMN short_channel_id bigint DEFAULT null", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates DROP COLUMN direction", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ADD COLUMN direction boolean DEFAULT null", &[]).await.unwrap();
+ loop {
+ let rows = tx.query("SELECT id, composite_index FROM channel_updates WHERE short_channel_id IS NULL LIMIT 50000", &[]).await.unwrap();
+ if rows.is_empty() { break; }
+ let mut updates = FuturesUnordered::new();
+ for row in rows {
+ let id: i32 = row.get("id");
+ let index: String = row.get("composite_index");
+ let tx_ref = &tx;
+ updates.push(async move {
+ let mut index_iter = index.split(":");
+ let scid_hex = index_iter.next().unwrap();
+ index_iter.next().unwrap();
+ let direction_str = index_iter.next().unwrap();
+ assert!(direction_str == "1" || direction_str == "0");
+ let direction = direction_str == "1";
+ let scid_be_bytes = hex_utils::to_vec(scid_hex).unwrap();
+ let scid = i64::from_be_bytes(scid_be_bytes.try_into().unwrap());
+ assert!(scid > 0); // Will roll over in some 150 years or so
+ tx_ref.execute("UPDATE channel_updates SET short_channel_id = $1, direction = $2 WHERE id = $3", &[&scid, &direction, &id]).await.unwrap();
+ });
+ }
+ while let Some(_) = updates.next().await {}
+ }
+ tx.execute("ALTER TABLE channel_updates ALTER short_channel_id DROP DEFAULT", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER short_channel_id SET NOT NULL", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER direction DROP DEFAULT", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER direction SET NOT NULL", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 3 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
+ if schema >= 1 && schema <= 3 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("ALTER TABLE channel_announcements DROP COLUMN short_channel_id", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_announcements ADD COLUMN short_channel_id bigint DEFAULT null", &[]).await.unwrap();
+ loop {
+ let rows = tx.query("SELECT id, announcement_signed FROM channel_announcements WHERE short_channel_id IS NULL LIMIT 10000", &[]).await.unwrap();
+ if rows.is_empty() { break; }
+ let mut updates = FuturesUnordered::new();
+ for row in rows {
+ let id: i32 = row.get("id");
+ let announcement: Vec<u8> = row.get("announcement_signed");
+ let tx_ref = &tx;
+ updates.push(async move {
+ let scid = ChannelAnnouncement::read(&mut Cursor::new(announcement)).unwrap().contents.short_channel_id as i64;
+ assert!(scid > 0); // Will roll over in some 150 years or so
+ tx_ref.execute("UPDATE channel_announcements SET short_channel_id = $1 WHERE id = $2", &[&scid, &id]).await.unwrap();
+ });
+ }
+ while let Some(_) = updates.next().await {}
+ }
+ tx.execute("ALTER TABLE channel_announcements ADD CONSTRAINT channel_announcements_short_channel_id_key UNIQUE (short_channel_id)", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_announcements ALTER short_channel_id DROP DEFAULT", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_announcements ALTER short_channel_id SET NOT NULL", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 4 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
+ if schema >= 1 && schema <= 4 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER composite_index SET DATA TYPE character(29)", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 5 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
+ if schema >= 1 && schema <= 5 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("ALTER TABLE channel_updates ALTER channel_flags SET DATA TYPE smallint", &[]).await.unwrap();
+ tx.execute("ALTER TABLE channel_announcements DROP COLUMN block_height", &[]).await.unwrap();
+ 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 <= 7 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channels_seen", &[]).await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_scid", &[]).await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_direction", &[]).await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_seen", &[]).await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_scid_seen", &[]).await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_scid_dir_seen", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 8 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
+ if schema >= 1 && schema <= 8 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_seen", &[]).await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_scid_seen", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 9 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
+ if schema >= 1 && schema <= 9 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_scid_dir_seen", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 10 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
+ if schema >= 1 && schema <= 10 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_id_with_scid_dir_blob", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 11 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
+ if schema >= 1 && schema <= 11 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_seen_with_id_direction_blob", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 12 WHERE id = 1", &[]).await.unwrap();
+ tx.commit().await.unwrap();
+ }
+ if schema >= 1 && schema <= 12 {
+ let tx = client.transaction().await.unwrap();
+ tx.execute("DROP INDEX IF EXISTS channel_updates_timestamp_desc", &[]).await.unwrap();
+ tx.execute("UPDATE config SET db_schema = 13 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);
+ }
+ // PostgreSQL (at least v13, but likely later versions as well) handles insert-only tables
+ // *very* poorly. After some number of inserts, it refuses to rely on indexes, assuming them to
+ // be possibly-stale, until a VACUUM happens. Thus, we set the vacuum factor really low here,
+ // pushing PostgreSQL to vacuum often.
+ // See https://www.cybertec-postgresql.com/en/postgresql-autovacuum-insert-only-tables/
+ let _ = client.execute("ALTER TABLE channel_updates SET ( autovacuum_vacuum_insert_scale_factor = 0.005 );", &[]).await;
+ let _ = client.execute("ALTER TABLE channel_announcements SET ( autovacuum_vacuum_insert_scale_factor = 0.005 );", &[]).await;
+}
+