spesmilo / electrumx / 123

pending completion

Build # 123

Build Type

Pull #65

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/spesmilo/electrumx/commit/<a class=hub.com/spesmilo/electrumx/commit/<a class="double-link" href="https://git"><a class=hub.com/spesmilo/electrumx/commit/<a class="double-link" href="https://git"><a class=hub.com/spesmilo/electrumx/commit/<a class="double-link" href="https://git"><a class=hub.com/spesmilo/electrumx/commit/<a class="double-link" href="https://git"><a class=hub.com/spesmilo/electrumx/commit/dfa771f4bc32e40621815c0c61eeb823f50eb5b7">dfa771f4b">&lt;a href=&quot;https://github.com/spesmilo/electrumx/commit/</a><a class="double-link" href="https://github.com/spesmilo/electrumx/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/spesmilo/electrumx/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/spesmilo/electrumx/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/spesmilo/electrumx/commit/dfa771f4bc32e40621815c0c61eeb823f50eb5b7">dfa771f4b</a><a href="https://github.com/spesmilo/electrumx/commit/dfa771f4bc32e40621815c0c61eeb823f50eb5b7">&lt;a href=&quot;https://github.com/spesmilo/electrumx/commit/dfa771f4bc32e40621815c0c61eeb823f50eb5b7&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/spesmilo/electrumx/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/spesmilo/electrumx/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/spesmilo/electrumx/commit/dfa771f4bc32e40621815c0c61eeb823f50eb5b7&amp;quot;&amp;gt;dfa771f4b&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/spesmilo/electrumx/commit/dfa771f4bc32e40621815c0c61eeb823f50eb5b7&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/spesmilo/electrumx/commit/dfa771f4bc32e40621815c0c61eeb823&amp;lt;/a&amp;gt;f50eb5</a>b7&quot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/spesmilo/electrumx/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/spesmilo/electrumx/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/spesmilo/electrumx/commit/78789d17f&... (continued)

Pull Request Pull Request #65: db: change history db schema

Run Details

116 of 116 new or added lines in 2 files covered. (100.0%)

5037 of 7895 relevant lines covered (63.8%)

0.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

13.43

/electrumx/server/history.py

# Copyright (c) 2016-2018, Neil Booth
# Copyright (c) 2017, the ElectrumX authors
#
# All rights reserved.
#
# See the file "LICENCE" for information about the copyright
# and warranty status of this software.

'''History by script hash (address).'''

import ast
import bisect
import time
from array import array
from collections import defaultdict
from typing import TYPE_CHECKING, Type, Optional

import electrumx.lib.util as util
from electrumx.lib.hash import HASHX_LEN, hash_to_hex_str
from electrumx.lib.util import (pack_be_uint16, pack_le_uint64,
                                unpack_be_uint16_from, unpack_le_uint64)

if TYPE_CHECKING:
    from electrumx.server.storage import Storage


TXNUM_LEN = 5
FLUSHID_LEN = 2


class History:

    DB_VERSIONS = (0, 1, 2)

    db: Optional['Storage']

    def __init__(self):
        self.logger = util.class_logger(__name__, self.__class__.__name__)
        self.unflushed = defaultdict(bytearray)
        self.unflushed_count = 0
        self.hist_db_tx_count = 0
        self.hist_db_tx_count_next = 0  # after next flush, next value for self.hist_db_tx_count
        self.db_version = max(self.DB_VERSIONS)
        self.upgrade_cursor = -1

        # Key: address_hashX + tx_num
        # Value: <null>
        self.db = None

    def open_db(
            self,
            *,
            db_class: Type['Storage'],
            for_sync: bool,
            utxo_db_tx_count: int,
    ) -> None:
        self.db = db_class('hist', for_sync)
        self.read_state()
        self.clear_excess(utxo_db_tx_count)

    def close_db(self):
        if self.db:
            self.db.close()
            self.db = None

    def read_state(self):
        state = self.db.get(b'state\0\0')
        if state:
            state = ast.literal_eval(state.decode())
            if not isinstance(state, dict):
                raise RuntimeError('failed reading state from history DB')
            self.db_version = state.get('db_version', 0)
            self.upgrade_cursor = state.get('upgrade_cursor', -1)
            self.hist_db_tx_count = state.get('hist_db_tx_count', 0)
            self.hist_db_tx_count_next = self.hist_db_tx_count

        if self.db_version not in self.DB_VERSIONS:
            msg = (f'your history DB version is {self.db_version} but '
                   f'this software only handles DB versions {self.DB_VERSIONS}')
            self.logger.error(msg)
            raise RuntimeError(msg)
        if self.db_version != max(self.DB_VERSIONS):
            self.upgrade_db()
        self.logger.info(f'history DB version: {self.db_version}')

    def clear_excess(self, utxo_db_tx_count: int) -> None:
        # self.hist_db_tx_count != utxo_db_tx_count might happen as
        # both DBs cannot be updated atomically
        # FIXME when advancing blocks, hist_db is flushed first, so its count can be higher;
        #       but when backing up (e.g. reorg), hist_db is flushed first as well,
        #       so its count can be lower?!
        #       Shouldn't we flush utxo_db first when backing up?
        if self.hist_db_tx_count <= utxo_db_tx_count:
            assert self.hist_db_tx_count == utxo_db_tx_count
            return

        self.logger.info('DB shut down uncleanly.  Scanning for '
                         'excess history flushes...')

        txnum_padding = bytes(8-TXNUM_LEN)
        keys = []
        for db_key, db_val in self.db.iterator(prefix=b''):
            tx_numb = db_key[HASHX_LEN:]
            tx_num, = unpack_le_uint64(tx_numb + txnum_padding)
            if tx_num >= utxo_db_tx_count:
                keys.append(db_key)

        self.logger.info(f'deleting {len(keys):,d} history entries')

        self.hist_db_tx_count = utxo_db_tx_count
        self.hist_db_tx_count_next = self.hist_db_tx_count
        with self.db.write_batch() as batch:
            for key in keys:
                batch.delete(key)
            self.write_state(batch)

        self.logger.info('deleted excess history entries')

    def write_state(self, batch):
        '''Write state to the history DB.'''
        state = {
            'hist_db_tx_count': self.hist_db_tx_count,
            'db_version': self.db_version,
            'upgrade_cursor': self.upgrade_cursor,
        }
        # History entries are not prefixed; the suffix \0\0 is just for legacy reasons
        batch.put(b'state\0\0', repr(state).encode())

    def add_unflushed(self, hashXs_by_tx, first_tx_num):
        unflushed = self.unflushed
        count = 0
        tx_num = None
        for tx_num, hashXs in enumerate(hashXs_by_tx, start=first_tx_num):
            tx_numb = pack_le_uint64(tx_num)[:TXNUM_LEN]
            hashXs = set(hashXs)
            for hashX in hashXs:
                unflushed[hashX] += tx_numb
            count += len(hashXs)
        self.unflushed_count += count
        if tx_num is not None:
            assert self.hist_db_tx_count_next + len(hashXs_by_tx) == tx_num + 1
            self.hist_db_tx_count_next = tx_num + 1

    def unflushed_memsize(self):
        return len(self.unflushed) * 180 + self.unflushed_count * TXNUM_LEN

    def assert_flushed(self):
        assert not self.unflushed

    def flush(self):
        start_time = time.monotonic()
        unflushed = self.unflushed
        chunks = util.chunks

        with self.db.write_batch() as batch:
            for hashX in sorted(unflushed):
                for tx_num in chunks(unflushed[hashX], TXNUM_LEN):
                    db_key = hashX + tx_num
                    batch.put(db_key, b'')
            self.hist_db_tx_count = self.hist_db_tx_count_next
            self.write_state(batch)

        count = len(unflushed)
        unflushed.clear()
        self.unflushed_count = 0

        if self.db.for_sync:
            elapsed = time.monotonic() - start_time
            self.logger.info(f'flushed history in {elapsed:.1f}s '
                             f'for {count:,d} addrs')

    def backup(self, hashXs, tx_count):
        self.assert_flushed()
        nremoves = 0
        txnum_padding = bytes(8-TXNUM_LEN)
        with self.db.write_batch() as batch:
            for hashX in sorted(hashXs):
                deletes = []
                for db_key, db_val in self.db.iterator(prefix=hashX, reverse=True):
                    tx_numb = db_key[HASHX_LEN:]
                    tx_num, = unpack_le_uint64(tx_numb + txnum_padding)
                    if tx_num >= tx_count:
                        nremoves += 1
                        deletes.append(db_key)
                    else:
                        break
                for key in deletes:
                    batch.delete(key)
            self.hist_db_tx_count = tx_count
            self.hist_db_tx_count_next = self.hist_db_tx_count
            self.write_state(batch)

        self.logger.info(f'backing up removed {nremoves:,d} history entries')

    def get_txnums(self, hashX, limit=1000):
        '''Generator that returns an unpruned, sorted list of tx_nums in the
        history of a hashX.  Includes both spending and receiving
        transactions.  By default yields at most 1000 entries.  Set
        limit to None to get them all.  '''
        limit = util.resolve_limit(limit)
        txnum_padding = bytes(8-TXNUM_LEN)
        for db_key, db_val in self.db.iterator(prefix=hashX):
            tx_numb = db_key[HASHX_LEN:]
            if limit == 0:
                return
            tx_num, = unpack_le_uint64(tx_numb + txnum_padding)
            yield tx_num
            limit -= 1

    #
    # DB upgrade
    #

    def upgrade_db(self):
        self.logger.info(f'history DB current version: {self.db_version}. '
                         f'latest is: {max(self.DB_VERSIONS)}')
        self.logger.info('Upgrading your history DB; this can take some time...')

        def convert_version_1():
            def upgrade_cursor(cursor):
                count = 0
                prefix = pack_be_uint16(cursor)
                key_len = HASHX_LEN + 2
                chunks = util.chunks
                with self.db.write_batch() as batch:
                    batch_put = batch.put
                    for key, hist in self.db.iterator(prefix=prefix):
                        # Ignore non-history entries
                        if len(key) != key_len:
                            continue
                        count += 1
                        hist = b''.join(item + b'\0' for item in chunks(hist, 4))
                        batch_put(key, hist)
                    self.upgrade_cursor = cursor
                    self.write_state(batch)
                return count

            last = time.monotonic()
            count = 0

            for cursor in range(self.upgrade_cursor + 1, 65536):
                count += upgrade_cursor(cursor)
                now = time.monotonic()
                if now > last + 10:
                    last = now
                    self.logger.info(f'history DB v0->v1: {count:,d} entries updated, '
                                     f'{cursor * 100 / 65536:.1f}% complete')

            self.db_version = 1
            self.upgrade_cursor = -1
            with self.db.write_batch() as batch:
                self.write_state(batch)
            self.logger.info('history DB upgraded to v1 successfully')

        def convert_version_2():
            # old schema:
            # Key: address_hashX + flush_id
            # Value: sorted "list" of tx_nums in history of hashX
            # -----
            # new schema:
            # Key: address_hashX + tx_num
            # Value: <null>

            def upgrade_cursor(cursor):
                count = 0
                prefix = pack_be_uint16(cursor)
                key_len = HASHX_LEN + 2
                chunks = util.chunks
                txnum_padding = bytes(8-TXNUM_LEN)
                with self.db.write_batch() as batch:
                    batch_put = batch.put
                    batch_delete = batch.delete
                    max_tx_num = 0
                    for db_key, db_val in self.db.iterator(prefix=prefix):
                        # Ignore non-history entries
                        if len(db_key) != key_len:
                            continue
                        count += 1
                        batch_delete(db_key)
                        hashX = db_key[:HASHX_LEN]
                        for tx_numb in chunks(db_val, 5):
                            batch_put(hashX + tx_numb, b'')
                            tx_num, = unpack_le_uint64(tx_numb + txnum_padding)
                            max_tx_num = max(max_tx_num, tx_num)
                    self.upgrade_cursor = cursor
                    self.hist_db_tx_count = max(self.hist_db_tx_count, max_tx_num + 1)
                    self.hist_db_tx_count_next = self.hist_db_tx_count
                    self.write_state(batch)
                return count

            last = time.monotonic()
            count = 0

            for cursor in range(self.upgrade_cursor + 1, 65536):
                count += upgrade_cursor(cursor)
                now = time.monotonic()
                if now > last + 10:
                    last = now
                    self.logger.info(f'history DB v1->v2: {count:,d} entries updated, '
                                     f'{cursor * 100 / 65536:.1f}% complete')

            self.db_version = 2
            self.upgrade_cursor = -1
            with self.db.write_batch() as batch:
                self.write_state(batch)
            self.logger.info('history DB upgraded to v2 successfully')

        if self.db_version == 0:
            convert_version_1()
        if self.db_version == 1:
            convert_version_2()
        self.db_version = max(self.DB_VERSIONS)

1	# Copyright (c) 2016-2018, Neil Booth
2	# Copyright (c) 2017, the ElectrumX authors
3	#
4	# All rights reserved.
5	#
6	# See the file "LICENCE" for information about the copyright
7	# and warranty status of this software.
8
9	'''History by script hash (address).'''	1✔
10
11	import ast	1✔
12	import bisect	1✔
13	import time	1✔
14	from array import array	1✔
15	from collections import defaultdict	1✔
16	from typing import TYPE_CHECKING, Type, Optional	1✔
17
18	import electrumx.lib.util as util	1✔
19	from electrumx.lib.hash import HASHX_LEN, hash_to_hex_str	1✔
20	from electrumx.lib.util import (pack_be_uint16, pack_le_uint64,	1✔
21	unpack_be_uint16_from, unpack_le_uint64)
22
23	if TYPE_CHECKING:	1✔
24	from electrumx.server.storage import Storage	×
25
26
27	TXNUM_LEN = 5	1✔
28	FLUSHID_LEN = 2	1✔
29
30
31	class History:	1✔
32
33	DB_VERSIONS = (0, 1, 2)	1✔
34
35	db: Optional['Storage']	1✔
36
37	def __init__(self):	1✔
38	self.logger = util.class_logger(__name__, self.__class__.__name__)	×
39	self.unflushed = defaultdict(bytearray)	×
40	self.unflushed_count = 0	×
41	self.hist_db_tx_count = 0	×
42	self.hist_db_tx_count_next = 0 # after next flush, next value for self.hist_db_tx_count	×
43	self.db_version = max(self.DB_VERSIONS)	×
44	self.upgrade_cursor = -1	×
45
46	# Key: address_hashX + tx_num
47	# Value: <null>
48	self.db = None	×
49
50	def open_db(	1✔
51	self,
52	*,
53	db_class: Type['Storage'],
54	for_sync: bool,
55	utxo_db_tx_count: int,
56	) -> None:
57	self.db = db_class('hist', for_sync)	×
58	self.read_state()	×
59	self.clear_excess(utxo_db_tx_count)	×
60
61	def close_db(self):	1✔
62	if self.db:	×
63	self.db.close()	×
64	self.db = None	×
65
66	def read_state(self):	1✔
67	state = self.db.get(b'state\0\0')	×
68	if state:	×
69	state = ast.literal_eval(state.decode())	×
70	if not isinstance(state, dict):	×
71	raise RuntimeError('failed reading state from history DB')	×
72	self.db_version = state.get('db_version', 0)	×
73	self.upgrade_cursor = state.get('upgrade_cursor', -1)	×
74	self.hist_db_tx_count = state.get('hist_db_tx_count', 0)	×
75	self.hist_db_tx_count_next = self.hist_db_tx_count	×
76
77	if self.db_version not in self.DB_VERSIONS:	×
78	msg = (f'your history DB version is {self.db_version} but '	×
79	f'this software only handles DB versions {self.DB_VERSIONS}')
80	self.logger.error(msg)	×
81	raise RuntimeError(msg)	×
82	if self.db_version != max(self.DB_VERSIONS):	×
83	self.upgrade_db()	×
84	self.logger.info(f'history DB version: {self.db_version}')	×
85
86	def clear_excess(self, utxo_db_tx_count: int) -> None:	1✔
87	# self.hist_db_tx_count != utxo_db_tx_count might happen as
88	# both DBs cannot be updated atomically
89	# FIXME when advancing blocks, hist_db is flushed first, so its count can be higher;
90	# but when backing up (e.g. reorg), hist_db is flushed first as well,
91	# so its count can be lower?!
92	# Shouldn't we flush utxo_db first when backing up?
93	if self.hist_db_tx_count <= utxo_db_tx_count:	×
94	assert self.hist_db_tx_count == utxo_db_tx_count	×
95	return	×
96
97	self.logger.info('DB shut down uncleanly. Scanning for '	×
98	'excess history flushes...')
99
100	txnum_padding = bytes(8-TXNUM_LEN)	×
101	keys = []	×
102	for db_key, db_val in self.db.iterator(prefix=b''):	×
103	tx_numb = db_key[HASHX_LEN:]	×
104	tx_num, = unpack_le_uint64(tx_numb + txnum_padding)	×
105	if tx_num >= utxo_db_tx_count:	×
106	keys.append(db_key)	×
107
108	self.logger.info(f'deleting {len(keys):,d} history entries')	×
109
110	self.hist_db_tx_count = utxo_db_tx_count	×
111	self.hist_db_tx_count_next = self.hist_db_tx_count	×
112	with self.db.write_batch() as batch:	×
113	for key in keys:	×
114	batch.delete(key)	×
115	self.write_state(batch)	×
116
117	self.logger.info('deleted excess history entries')	×
118
119	def write_state(self, batch):	1✔
120	'''Write state to the history DB.'''
121	state = {	×
122	'hist_db_tx_count': self.hist_db_tx_count,
123	'db_version': self.db_version,
124	'upgrade_cursor': self.upgrade_cursor,
125	}
126	# History entries are not prefixed; the suffix \0\0 is just for legacy reasons
127	batch.put(b'state\0\0', repr(state).encode())	×
128
129	def add_unflushed(self, hashXs_by_tx, first_tx_num):	1✔
130	unflushed = self.unflushed	×
131	count = 0	×
132	tx_num = None	×
133	for tx_num, hashXs in enumerate(hashXs_by_tx, start=first_tx_num):	×
134	tx_numb = pack_le_uint64(tx_num)[:TXNUM_LEN]	×
135	hashXs = set(hashXs)	×
136	for hashX in hashXs:	×
137	unflushed[hashX] += tx_numb	×
138	count += len(hashXs)	×
139	self.unflushed_count += count	×
140	if tx_num is not None:	×
141	assert self.hist_db_tx_count_next + len(hashXs_by_tx) == tx_num + 1	×
142	self.hist_db_tx_count_next = tx_num + 1	×
143
144	def unflushed_memsize(self):	1✔
145	return len(self.unflushed) * 180 + self.unflushed_count * TXNUM_LEN	×
146
147	def assert_flushed(self):	1✔
148	assert not self.unflushed	×
149
150	def flush(self):	1✔
151	start_time = time.monotonic()	×
152	unflushed = self.unflushed	×
153	chunks = util.chunks	×
154
155	with self.db.write_batch() as batch:	×
156	for hashX in sorted(unflushed):	×
157	for tx_num in chunks(unflushed[hashX], TXNUM_LEN):	×
158	db_key = hashX + tx_num	×
159	batch.put(db_key, b'')	×
160	self.hist_db_tx_count = self.hist_db_tx_count_next	×
161	self.write_state(batch)	×
162
163	count = len(unflushed)	×
164	unflushed.clear()	×
165	self.unflushed_count = 0	×
166
167	if self.db.for_sync:	×
168	elapsed = time.monotonic() - start_time	×
169	self.logger.info(f'flushed history in {elapsed:.1f}s '	×
170	f'for {count:,d} addrs')
171
172	def backup(self, hashXs, tx_count):	1✔
173	self.assert_flushed()	×
174	nremoves = 0	×
175	txnum_padding = bytes(8-TXNUM_LEN)	×
176	with self.db.write_batch() as batch:	×
177	for hashX in sorted(hashXs):	×
178	deletes = []	×
179	for db_key, db_val in self.db.iterator(prefix=hashX, reverse=True):	×
180	tx_numb = db_key[HASHX_LEN:]	×
181	tx_num, = unpack_le_uint64(tx_numb + txnum_padding)	×
182	if tx_num >= tx_count:	×
183	nremoves += 1	×
184	deletes.append(db_key)	×
185	else:
186	break	×
187	for key in deletes:	×
188	batch.delete(key)	×
189	self.hist_db_tx_count = tx_count	×
190	self.hist_db_tx_count_next = self.hist_db_tx_count	×
191	self.write_state(batch)	×
192
193	self.logger.info(f'backing up removed {nremoves:,d} history entries')	×
194
195	def get_txnums(self, hashX, limit=1000):	1✔
196	'''Generator that returns an unpruned, sorted list of tx_nums in the
197	history of a hashX. Includes both spending and receiving
198	transactions. By default yields at most 1000 entries. Set
199	limit to None to get them all. '''
200	limit = util.resolve_limit(limit)	×
201	txnum_padding = bytes(8-TXNUM_LEN)	×
202	for db_key, db_val in self.db.iterator(prefix=hashX):	×
203	tx_numb = db_key[HASHX_LEN:]	×
204	if limit == 0:	×
205	return	×
206	tx_num, = unpack_le_uint64(tx_numb + txnum_padding)	×
207	yield tx_num	×
208	limit -= 1	×
209
210	#
211	# DB upgrade
212	#
213
214	def upgrade_db(self):	1✔
215	self.logger.info(f'history DB current version: {self.db_version}. '	×
216	f'latest is: {max(self.DB_VERSIONS)}')
217	self.logger.info('Upgrading your history DB; this can take some time...')	×
218
219	def convert_version_1():	×
220	def upgrade_cursor(cursor):	×
221	count = 0	×
222	prefix = pack_be_uint16(cursor)	×
223	key_len = HASHX_LEN + 2	×
224	chunks = util.chunks	×
225	with self.db.write_batch() as batch:	×
226	batch_put = batch.put	×
227	for key, hist in self.db.iterator(prefix=prefix):	×
228	# Ignore non-history entries
229	if len(key) != key_len:	×
230	continue	×
231	count += 1	×
232	hist = b''.join(item + b'\0' for item in chunks(hist, 4))	×
233	batch_put(key, hist)	×
234	self.upgrade_cursor = cursor	×
235	self.write_state(batch)	×
236	return count	×
237
238	last = time.monotonic()	×
239	count = 0	×
240
241	for cursor in range(self.upgrade_cursor + 1, 65536):	×
242	count += upgrade_cursor(cursor)	×
243	now = time.monotonic()	×
244	if now > last + 10:	×
245	last = now	×
246	self.logger.info(f'history DB v0->v1: {count:,d} entries updated, '	×
247	f'{cursor * 100 / 65536:.1f}% complete')
248
249	self.db_version = 1	×
250	self.upgrade_cursor = -1	×
251	with self.db.write_batch() as batch:	×
252	self.write_state(batch)	×
253	self.logger.info('history DB upgraded to v1 successfully')	×
254
255	def convert_version_2():	×
256	# old schema:
257	# Key: address_hashX + flush_id
258	# Value: sorted "list" of tx_nums in history of hashX
259	# -----
260	# new schema:
261	# Key: address_hashX + tx_num
262	# Value: <null>
263
264	def upgrade_cursor(cursor):	×
265	count = 0	×
266	prefix = pack_be_uint16(cursor)	×
267	key_len = HASHX_LEN + 2	×
268	chunks = util.chunks	×
269	txnum_padding = bytes(8-TXNUM_LEN)	×
270	with self.db.write_batch() as batch:	×
271	batch_put = batch.put	×
272	batch_delete = batch.delete	×
273	max_tx_num = 0	×
274	for db_key, db_val in self.db.iterator(prefix=prefix):	×
275	# Ignore non-history entries
276	if len(db_key) != key_len:	×
277	continue	×
278	count += 1	×
279	batch_delete(db_key)	×
280	hashX = db_key[:HASHX_LEN]	×
281	for tx_numb in chunks(db_val, 5):	×
282	batch_put(hashX + tx_numb, b'')	×
283	tx_num, = unpack_le_uint64(tx_numb + txnum_padding)	×
284	max_tx_num = max(max_tx_num, tx_num)	×
285	self.upgrade_cursor = cursor	×
286	self.hist_db_tx_count = max(self.hist_db_tx_count, max_tx_num + 1)	×
287	self.hist_db_tx_count_next = self.hist_db_tx_count	×
288	self.write_state(batch)	×
289	return count	×
290
291	last = time.monotonic()	×
292	count = 0	×
293
294	for cursor in range(self.upgrade_cursor + 1, 65536):	×
295	count += upgrade_cursor(cursor)	×
296	now = time.monotonic()	×
297	if now > last + 10:	×
298	last = now	×
299	self.logger.info(f'history DB v1->v2: {count:,d} entries updated, '	×
300	f'{cursor * 100 / 65536:.1f}% complete')
301
302	self.db_version = 2	×
303	self.upgrade_cursor = -1	×
304	with self.db.write_batch() as batch:	×
305	self.write_state(batch)	×
306	self.logger.info('history DB upgraded to v2 successfully')	×
307
308	if self.db_version == 0:	×
309	convert_version_1()	×
310	if self.db_version == 1:	×
311	convert_version_2()	×
312	self.db_version = max(self.DB_VERSIONS)	×

spesmilo / electrumx / 123

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous