forked from LBRYCommunity/lbry-sdk
295 lines
12 KiB
Python
295 lines
12 KiB
Python
import logging
|
|
from typing import Tuple
|
|
|
|
from sqlalchemy import table, text, func, union, between
|
|
from sqlalchemy.future import select
|
|
from sqlalchemy.schema import CreateTable
|
|
|
|
from lbry.db.tables import (
|
|
Block as BlockTable, BlockFilter, BlockGroupFilter,
|
|
TX, TXFilter, MempoolFilter, TXO, TXI, Claim, Tag, Support
|
|
)
|
|
from lbry.db.tables import (
|
|
pg_add_block_constraints_and_indexes,
|
|
pg_add_block_filter_constraints_and_indexes,
|
|
pg_add_tx_constraints_and_indexes,
|
|
pg_add_tx_filter_constraints_and_indexes,
|
|
pg_add_txo_constraints_and_indexes,
|
|
pg_add_txi_constraints_and_indexes,
|
|
)
|
|
from lbry.db.query_context import ProgressContext, event_emitter, context
|
|
from lbry.db.sync import set_input_addresses, update_spent_outputs
|
|
from lbry.blockchain.block import Block, create_address_filter
|
|
from lbry.blockchain.bcd_data_stream import BCDataStream
|
|
|
|
from .context import get_or_initialize_lbrycrd
|
|
from .filter_builder import FilterBuilder
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def get_best_block_height_for_file(file_number):
|
|
return context().fetchone(
|
|
select(func.coalesce(func.max(BlockTable.c.height), -1).label('height'))
|
|
.where(BlockTable.c.file_number == file_number)
|
|
)['height']
|
|
|
|
|
|
@event_emitter("blockchain.sync.blocks.file", "blocks", "txs", throttle=100)
|
|
def sync_block_file(
|
|
file_number: int, start_height: int, txs: int, flush_size: int, p: ProgressContext
|
|
):
|
|
chain = get_or_initialize_lbrycrd(p.ctx)
|
|
new_blocks = chain.db.sync_get_blocks_in_file(file_number, start_height)
|
|
if not new_blocks:
|
|
return -1
|
|
file_name = chain.get_block_file_name(file_number)
|
|
p.start(len(new_blocks), txs, progress_id=file_number, label=file_name)
|
|
block_file_path = chain.get_block_file_path(file_number)
|
|
done_blocks = done_txs = 0
|
|
last_block_processed, loader = -1, p.ctx.get_bulk_loader()
|
|
with open(block_file_path, "rb") as fp:
|
|
stream = BCDataStream(fp=fp)
|
|
for done_blocks, block_info in enumerate(new_blocks, start=1):
|
|
block_height = block_info["height"]
|
|
fp.seek(block_info["data_offset"])
|
|
block = Block.from_data_stream(stream, block_height, file_number)
|
|
loader.add_block(block)
|
|
if len(loader.txs) >= flush_size:
|
|
done_txs += loader.flush(TX)
|
|
p.step(done_blocks, done_txs)
|
|
last_block_processed = block_height
|
|
if p.ctx.stop_event.is_set():
|
|
return last_block_processed
|
|
if loader.txs:
|
|
done_txs += loader.flush(TX)
|
|
p.step(done_blocks, done_txs)
|
|
return last_block_processed
|
|
|
|
|
|
@event_emitter("blockchain.sync.blocks.indexes", "steps")
|
|
def blocks_constraints_and_indexes(p: ProgressContext):
|
|
p.start(1 + len(pg_add_block_constraints_and_indexes))
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM ANALYZE block;"))
|
|
p.step()
|
|
for constraint in pg_add_block_constraints_and_indexes:
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute(text(constraint))
|
|
p.step()
|
|
|
|
|
|
@event_emitter("blockchain.sync.blocks.vacuum", "steps")
|
|
def blocks_vacuum(p: ProgressContext):
|
|
p.start(1)
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM block;"))
|
|
p.step()
|
|
|
|
|
|
@event_emitter("blockchain.sync.spends.main", "steps")
|
|
def sync_spends(initial_sync: bool, p: ProgressContext):
|
|
if initial_sync:
|
|
p.start(
|
|
7 +
|
|
len(pg_add_tx_constraints_and_indexes) +
|
|
len(pg_add_txi_constraints_and_indexes) +
|
|
len(pg_add_txo_constraints_and_indexes)
|
|
)
|
|
# 1. tx table stuff
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM ANALYZE tx;"))
|
|
p.step()
|
|
for constraint in pg_add_tx_constraints_and_indexes:
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute(text(constraint))
|
|
p.step()
|
|
# A. Update TXIs to have the address of TXO they are spending.
|
|
# 2. txi table reshuffling
|
|
p.ctx.execute(text("ALTER TABLE txi RENAME TO old_txi;"))
|
|
p.ctx.execute(CreateTable(TXI, include_foreign_key_constraints=[]))
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute(text("ALTER TABLE txi DROP CONSTRAINT txi_pkey;"))
|
|
p.step()
|
|
# 3. insert
|
|
old_txi = table("old_txi", *(c.copy() for c in TXI.columns)) # pylint: disable=not-an-iterable
|
|
columns = [c for c in old_txi.columns if c.name != "address"] + [TXO.c.address]
|
|
join_txi_on_txo = old_txi.join(TXO, old_txi.c.txo_hash == TXO.c.txo_hash)
|
|
select_txis = select(*columns).select_from(join_txi_on_txo)
|
|
insert_txis = TXI.insert().from_select(columns, select_txis)
|
|
p.ctx.execute(insert_txis)
|
|
p.step()
|
|
# 4. drop old txi and vacuum
|
|
p.ctx.execute(text("DROP TABLE old_txi;"))
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM ANALYZE txi;"))
|
|
p.step()
|
|
for constraint in pg_add_txi_constraints_and_indexes:
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute(text(constraint))
|
|
p.step()
|
|
# B. Update TXOs to have the height at which they were spent (if they were).
|
|
# 5. txo table reshuffling
|
|
p.ctx.execute(text("ALTER TABLE txo RENAME TO old_txo;"))
|
|
p.ctx.execute(CreateTable(TXO, include_foreign_key_constraints=[]))
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute(text("ALTER TABLE txo DROP CONSTRAINT txo_pkey;"))
|
|
p.step()
|
|
# 6. insert
|
|
old_txo = table("old_txo", *(c.copy() for c in TXO.columns)) # pylint: disable=not-an-iterable
|
|
columns = [c for c in old_txo.columns if c.name != "spent_height"]
|
|
insert_columns = columns + [TXO.c.spent_height]
|
|
select_columns = columns + [func.coalesce(TXI.c.height, 0).label("spent_height")]
|
|
join_txo_on_txi = old_txo.join(TXI, old_txo.c.txo_hash == TXI.c.txo_hash, isouter=True)
|
|
select_txos = select(*select_columns).select_from(join_txo_on_txi)
|
|
insert_txos = TXO.insert().from_select(insert_columns, select_txos)
|
|
p.ctx.execute(insert_txos)
|
|
p.step()
|
|
# 7. drop old txo
|
|
p.ctx.execute(text("DROP TABLE old_txo;"))
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM ANALYZE txo;"))
|
|
p.step()
|
|
for constraint in pg_add_txo_constraints_and_indexes:
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute(text(constraint))
|
|
p.step()
|
|
else:
|
|
p.start(5)
|
|
# 1. Update spent TXOs setting spent_height
|
|
update_spent_outputs(p.ctx)
|
|
p.step()
|
|
# 2. Update TXIs to have the address of TXO they are spending.
|
|
set_input_addresses(p.ctx)
|
|
p.step()
|
|
# 3. Update tx visibility map, which speeds up index-only scans.
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM tx;"))
|
|
p.step()
|
|
# 4. Update txi visibility map, which speeds up index-only scans.
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM txi;"))
|
|
p.step()
|
|
# 4. Update txo visibility map, which speeds up index-only scans.
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM txo;"))
|
|
p.step()
|
|
|
|
|
|
@event_emitter("blockchain.sync.filters.generate", "blocks", throttle=100)
|
|
def sync_filters(start, end, p: ProgressContext):
|
|
fp = FilterBuilder(start, end)
|
|
p.start((end-start)+1, progress_id=start, label=f"generate filters {start}-{end}")
|
|
with p.ctx.connect_streaming() as c:
|
|
loader = p.ctx.get_bulk_loader()
|
|
|
|
tx_hash, height, addresses, last_added = None, None, set(), None
|
|
address_to_hash = p.ctx.ledger.address_to_hash160
|
|
for row in c.execute(get_block_tx_addresses_sql(*fp.query_heights)):
|
|
if tx_hash != row.tx_hash:
|
|
if tx_hash is not None:
|
|
last_added = tx_hash
|
|
fp.add(tx_hash, height, addresses)
|
|
tx_hash, height, addresses = row.tx_hash, row.height, set()
|
|
addresses.add(address_to_hash(row.address))
|
|
if all([last_added, tx_hash]) and last_added != tx_hash: # pickup last tx
|
|
fp.add(tx_hash, height, addresses)
|
|
|
|
for tx_hash, height, addresses in fp.tx_filters:
|
|
loader.add_transaction_filter(
|
|
tx_hash, height, create_address_filter(list(addresses))
|
|
)
|
|
|
|
for height, addresses in fp.block_filters.items():
|
|
loader.add_block_filter(
|
|
height, create_address_filter(list(addresses))
|
|
)
|
|
|
|
for group_filter in fp.group_filters:
|
|
for height, addresses in group_filter.groups.items():
|
|
loader.add_group_filter(
|
|
height, group_filter.factor, create_address_filter(list(addresses))
|
|
)
|
|
|
|
p.add(loader.flush(BlockFilter))
|
|
|
|
|
|
@event_emitter("blockchain.sync.filters.indexes", "steps")
|
|
def filters_constraints_and_indexes(p: ProgressContext):
|
|
constraints = (
|
|
pg_add_tx_filter_constraints_and_indexes +
|
|
pg_add_block_filter_constraints_and_indexes
|
|
)
|
|
p.start(2 + len(constraints))
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM ANALYZE block_filter;"))
|
|
p.step()
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM ANALYZE tx_filter;"))
|
|
p.step()
|
|
for constraint in constraints:
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute(text(constraint))
|
|
p.step()
|
|
|
|
|
|
@event_emitter("blockchain.sync.filters.vacuum", "steps")
|
|
def filters_vacuum(p: ProgressContext):
|
|
p.start(2)
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM block_filter;"))
|
|
p.step()
|
|
if p.ctx.is_postgres:
|
|
p.ctx.execute_notx(text("VACUUM tx_filter;"))
|
|
p.step()
|
|
|
|
|
|
def get_block_range_without_filters() -> Tuple[int, int]:
|
|
sql = (
|
|
select(
|
|
func.coalesce(func.min(BlockTable.c.height), -1).label('start_height'),
|
|
func.coalesce(func.max(BlockTable.c.height), -1).label('end_height'),
|
|
)
|
|
.select_from(
|
|
BlockTable.join(BlockFilter, BlockTable.c.height == BlockFilter.c.height, isouter=True)
|
|
)
|
|
.where(BlockFilter.c.height == None)
|
|
)
|
|
result = context().fetchone(sql)
|
|
return result['start_height'], result['end_height']
|
|
|
|
|
|
def get_block_tx_addresses_sql(start_height, end_height):
|
|
return union(
|
|
select(TXO.c.tx_hash, TXO.c.height, TXO.c.address).where(
|
|
(TXO.c.address.isnot(None)) & between(TXO.c.height, start_height, end_height)
|
|
),
|
|
select(TXI.c.tx_hash, TXI.c.height, TXI.c.address).where(
|
|
(TXI.c.address.isnot(None)) & between(TXI.c.height, start_height, end_height)
|
|
),
|
|
).order_by('height', 'tx_hash')
|
|
|
|
|
|
@event_emitter("blockchain.sync.rewind.main", "steps")
|
|
def rewind(height: int, p: ProgressContext):
|
|
deletes = [
|
|
BlockTable.delete().where(BlockTable.c.height >= height),
|
|
TXI.delete().where(TXI.c.height >= height),
|
|
TXO.delete().where(TXO.c.height >= height),
|
|
TX.delete().where(TX.c.height >= height),
|
|
Tag.delete().where(
|
|
Tag.c.claim_hash.in_(
|
|
select(Claim.c.claim_hash).where(Claim.c.height >= height)
|
|
)
|
|
),
|
|
Claim.delete().where(Claim.c.height >= height),
|
|
Support.delete().where(Support.c.height >= height),
|
|
BlockFilter.delete().where(BlockFilter.c.height >= height),
|
|
# TODO: group and tx filters need where() clauses (below actually breaks things)
|
|
BlockGroupFilter.delete(),
|
|
TXFilter.delete(),
|
|
MempoolFilter.delete()
|
|
]
|
|
for delete in p.iter(deletes):
|
|
p.ctx.execute(delete)
|