132 lines
5.1 KiB
Python
132 lines
5.1 KiB
Python
import asyncio
|
|
import socket
|
|
import time
|
|
import logging
|
|
from unittest.mock import Mock
|
|
from torba.testcase import IntegrationTestCase, Conductor
|
|
import lbry.wallet
|
|
from lbry.schema.claim import Claim
|
|
from lbry.wallet.transaction import Transaction, Output
|
|
from lbry.wallet.dewies import dewies_to_lbc as d2l, lbc_to_dewies as l2d
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
def wrap_callback_event(fn, callback):
|
|
def inner(*a, **kw):
|
|
callback()
|
|
return fn(*a, **kw)
|
|
return inner
|
|
|
|
|
|
class TestSessionBloat(IntegrationTestCase):
|
|
"""
|
|
ERROR:asyncio:Fatal read error on socket transport
|
|
protocol: <lbrynet.wallet.server.session.LBRYElectrumX object at 0x7f7e3bfcaf60>
|
|
transport: <_SelectorSocketTransport fd=3236 read=polling write=<idle, bufsize=0>>
|
|
Traceback (most recent call last):
|
|
File "/usr/lib/python3.7/asyncio/selector_events.py", line 801, in _read_ready__data_received
|
|
data = self._sock.recv(self.max_size)
|
|
TimeoutError: [Errno 110] Connection timed out
|
|
"""
|
|
|
|
LEDGER = lbry.wallet
|
|
|
|
async def asyncSetUp(self):
|
|
self.conductor = Conductor(
|
|
ledger_module=self.LEDGER, manager_module=self.MANAGER, verbosity=self.VERBOSITY
|
|
)
|
|
await self.conductor.start_blockchain()
|
|
self.addCleanup(self.conductor.stop_blockchain)
|
|
|
|
await self.conductor.start_spv()
|
|
|
|
self.session_manager = self.conductor.spv_node.server.session_mgr
|
|
self.session_manager.servers['TCP'].sockets[0].setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 64)
|
|
self.session_manager.servers['TCP'].sockets[0].setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 64)
|
|
|
|
self.addCleanup(self.conductor.stop_spv)
|
|
await self.conductor.start_wallet()
|
|
self.addCleanup(self.conductor.stop_wallet)
|
|
|
|
self.client_session = list(self.session_manager.sessions)[0]
|
|
self.client_session.transport.set_write_buffer_limits(0, 0)
|
|
|
|
self.paused_session = asyncio.Event(loop=self.loop)
|
|
self.resumed_session = asyncio.Event(loop=self.loop)
|
|
|
|
def paused():
|
|
self.resumed_session.clear()
|
|
self.paused_session.set()
|
|
|
|
def delayed_resume():
|
|
self.paused_session.clear()
|
|
|
|
time.sleep(1)
|
|
self.resumed_session.set()
|
|
|
|
self.client_session.pause_writing = wrap_callback_event(self.client_session.pause_writing, paused)
|
|
self.client_session.resume_writing = wrap_callback_event(self.client_session.resume_writing, delayed_resume)
|
|
|
|
self.blockchain = self.conductor.blockchain_node
|
|
self.wallet_node = self.conductor.wallet_node
|
|
self.manager = self.wallet_node.manager
|
|
self.ledger = self.wallet_node.ledger
|
|
self.wallet = self.wallet_node.wallet
|
|
self.account = self.wallet_node.wallet.default_account
|
|
|
|
async def test_session_bloat_from_socket_timeout(self):
|
|
await self.account.ensure_address_gap()
|
|
|
|
address1, address2 = await self.account.receiving.get_addresses(limit=2, only_usable=True)
|
|
sendtxid1 = await self.blockchain.send_to_address(address1, 5)
|
|
sendtxid2 = await self.blockchain.send_to_address(address2, 5)
|
|
|
|
await self.blockchain.generate(1)
|
|
await asyncio.wait([
|
|
self.on_transaction_id(sendtxid1),
|
|
self.on_transaction_id(sendtxid2)
|
|
])
|
|
|
|
self.assertEqual(d2l(await self.account.get_balance()), '10.0')
|
|
|
|
channel = Claim()
|
|
channel_txo = Output.pay_claim_name_pubkey_hash(
|
|
l2d('1.0'), '@bar', channel, self.account.ledger.address_to_hash160(address1)
|
|
)
|
|
channel_txo.generate_channel_private_key()
|
|
channel_txo.script.generate()
|
|
channel_tx = await Transaction.create([], [channel_txo], [self.account], self.account)
|
|
|
|
stream = Claim()
|
|
stream.stream.description = "0" * 8000
|
|
stream_txo = Output.pay_claim_name_pubkey_hash(
|
|
l2d('1.0'), 'foo', stream, self.account.ledger.address_to_hash160(address1)
|
|
)
|
|
stream_tx = await Transaction.create([], [stream_txo], [self.account], self.account)
|
|
stream_txo.sign(channel_txo)
|
|
await stream_tx.sign([self.account])
|
|
self.paused_session.clear()
|
|
self.resumed_session.clear()
|
|
|
|
await self.broadcast(channel_tx)
|
|
await self.broadcast(stream_tx)
|
|
await asyncio.wait_for(self.paused_session.wait(), 2)
|
|
self.assertEqual(1, len(self.session_manager.sessions))
|
|
|
|
real_sock = self.client_session.transport._extra.pop('socket')
|
|
mock_sock = Mock(spec=socket.socket)
|
|
|
|
for attr in dir(real_sock):
|
|
if not attr.startswith('__'):
|
|
setattr(mock_sock, attr, getattr(real_sock, attr))
|
|
|
|
def recv(*a, **kw):
|
|
raise TimeoutError("[Errno 110] Connection timed out")
|
|
|
|
mock_sock.recv = recv
|
|
self.client_session.transport._sock = mock_sock
|
|
self.client_session.transport._extra['socket'] = mock_sock
|
|
self.assertFalse(self.resumed_session.is_set())
|
|
self.assertFalse(self.session_manager.session_event.is_set())
|
|
await self.session_manager.session_event.wait()
|
|
self.assertEqual(0, len(self.session_manager.sessions))
|