# Copyright (c) The btclib developers
# Distributed under the MIT software license, see the accompanying
# LICENSE file or https://opensource.org/license/mit for the full text.
"""Where a peer is, and the table of the ones this node knows of.
The address itself is btclib's, and `btclib.p2p.addrv2.NetworkAddressV2`
is the one this node holds a peer in: BIP155's record is the only
encoding that carries every network a peer can be on, so the narrower
`addr` entry would lose an onion peer the moment one is gossiped. What
goes on the wire is that entry all the same wherever the peer has not
asked for BIP155, and `addr_entry` and `peer_from_addr_entry` are the
translation.
What is left here is what btclib has no business holding: dialling a
socket, and the table of addresses to dial. btclib is a codec -- it
speaks to nobody -- so the question "can this be connected to" and the
answer to it are this node's.
"""
import asyncio
import secrets
import socket
import threading
import time
from contextlib import contextmanager
from dataclasses import replace
from ipaddress import IPv4Address, IPv6Address, ip_address
from typing import TYPE_CHECKING, cast
from btclib.p2p.address import NetworkAddress, ServiceFlags, TimestampedNetworkAddress
from btclib.p2p.addrv2 import BIP155Network, NetworkAddressV2
from btclib_node.db import KeyValueStore
from btclib_node.exceptions import UnsupportedAddressTypeError
if TYPE_CHECKING:
from collections.abc import Iterable, Iterator
from pathlib import Path
from btclib_node.chains import Chain
__all__ = [
"PeerDB",
"addr_entry",
"can_addrv1",
"can_connect",
"dial",
"endpoint_key",
"ip_and_port",
"network_address",
"peer_address",
"peer_from_addr_entry",
]
# the two ids whose address field is an IP address, which is the whole
# of what an addr version 1 entry can carry, and the whole of what
# `dial` below opens a socket for
_IP_NETWORKS = (BIP155Network.IPV4, BIP155Network.IPV6)
# BIP155: a client SHOULD ignore an IPV6 entry whose sixteen octets fall
# in a range reserved for embedding another network's address into an
# IPv6 one -- the IPv4 mapping, `::ffff:0:0/96`, and OnionCat's
# `fd87:d87e:eb43::/48`, once used to carry a TORv2 address the same way.
# Core's `netaddress.h` (58a7869f86) is `IPV4_IN_IPV6_PREFIX` and
# `TORV2_IN_IPV6_PREFIX`; `IPv6Address.ipv4_mapped` is `ipaddress`'s own
# name for the first, and the second has no name of its own to borrow.
_ONIONCAT_PREFIX = b"\xfd\x87\xd8\x7e\xeb\x43"
[docs]
def peer_address(
ip: str, port: int, timestamp: int = 0, services: int = 0
) -> NetworkAddressV2:
"""Return the BIP155 record of a peer named by the text of its IP.
`ipaddress.ip_address` is what tells the two IP networks apart, and
the octets it packs are what BIP155 asks for: four for a v4 peer and
sixteen for a v6 one, where an `addr` entry would carry the v4 one
mapped into sixteen.
"""
parsed = ip_address(ip)
# `ip_address(...).version` is the stdlib's own name for this, not
# this tree's: it is only ever 4 or 6, and naming the 4 here would
# give a second name to something `ipaddress` already names by being
# IPv4 itself
network_id = (
BIP155Network.IPV4 if parsed.version == 4 else BIP155Network.IPV6 # noqa: PLR2004
)
return NetworkAddressV2(timestamp, services, network_id, parsed.packed, port)
[docs]
def can_addrv1(address: NetworkAddressV2) -> bool:
"""Answer whether an addr version 1 message has room for this peer."""
return address.network_id in _IP_NETWORKS
[docs]
def can_connect(address: NetworkAddressV2) -> bool:
"""Answer whether this node has a dial for the peer's network."""
return address.network_id in _IP_NETWORKS
def _is_embedded_ipv6(address: NetworkAddressV2) -> bool:
"""Answer whether an `IPV6` record's octets are really another network's.
The two BIP155 ignore rules `_ONIONCAT_PREFIX` documents, applied
together: `PeerDB.add_addresses` is where this is asked, that being
the point a record kept becomes an entry gossiped back to the next
peer -- `btclib.p2p.addrv2`'s own docstring calls both rules receive
policy and leaves them to the caller rather than the parser.
"""
return address.network_id == BIP155Network.IPV6 and (
IPv6Address(address.address).ipv4_mapped is not None
or address.address.startswith(_ONIONCAT_PREFIX)
)
[docs]
def network_address(address: NetworkAddressV2) -> NetworkAddress:
"""Return the untimestamped form of a BIP155 record, where it has one.
What a `version` message's two addresses are, and what an `addr`
entry is built on. `can_addrv1` is the question a caller asks first;
the refusal below is what makes the answer binding rather than
advisory, because the length would not catch it: BIP155 gives cjdns
and yggdrasil the sixteen octets an IPv6 address has, so `IPv6Address`
would take either for an IP address and hand back a peer that is not
the one that was gossiped.
"""
if not can_addrv1(address):
err_msg = f"not an ip address: network id {int(address.network_id)}"
raise ValueError(err_msg)
ip = (
IPv4Address(address.address)
if address.network_id == BIP155Network.IPV4
else IPv6Address(address.address)
)
return NetworkAddress(address.services, ip, address.port)
[docs]
def addr_entry(address: NetworkAddressV2) -> TimestampedNetworkAddress:
"""Return the addr version 1 entry a BIP155 record is, where it is one."""
return TimestampedNetworkAddress(address.timestamp, network_address(address))
[docs]
def peer_from_addr_entry(entry: TimestampedNetworkAddress) -> NetworkAddressV2:
"""Return the BIP155 record an addr version 1 entry describes.
An `addr` entry holds every address in sixteen octets, a v4 one
mapped into them, where BIP155 gives the two networks different ids
and different lengths: `ipv4_mapped` is what tells them apart, and it
is the reason this is not a field rename.
"""
ip = entry.address.ip
mapped = ip.ipv4_mapped
network_id = BIP155Network.IPV4 if mapped else BIP155Network.IPV6
return NetworkAddressV2(
entry.timestamp,
entry.address.services,
network_id,
mapped.packed if mapped else ip.packed,
entry.address.port,
)
[docs]
def ip_and_port(ip: str, port: int) -> str:
"""Return the endpoint the way Core's `CService::ToStringAddrPort` does.
`"[" + ToStringAddr() + "]:" + port_str` for every network that
function's `IsIPv4() || IsTor() || IsI2P() || IsInternal()` does not
name. The brackets are what tells a v6 host from its port:
`2001:db8::1` on port 8333 and `2001:db8::1:8333` on some other port
are both addresses, and without brackets both render as the second.
The host's text rather than the `NetworkAddress` a peer is held in,
because a socket's `getpeername` has no such object to offer and
answers with this.
A v4-mapped host is unwrapped rather than bracketed, which is Core's
answer too: `CNetAddr::SetLegacyIPv6` files a mapped address under
NET_IPV4, which that predicate names. Without the unwrapping a v4
peer would read `[::ffff:1.2.3.4]:8333`, a `NetworkAddress` holding
every address in the sixteen octets of an IPv6 one.
Raises `ValueError` where the host is not an IP address, which is
what `ipaddress.ip_address` answers with: a hostname is refused
rather than shown with brackets guessed at.
"""
parsed = ip_address(ip)
if not isinstance(parsed, IPv6Address):
return f"{parsed}:{port}"
mapped = parsed.ipv4_mapped
if mapped:
return f"{mapped}:{port}"
return f"[{parsed}]:{port}"
# Core's own default (`DEFAULT_CONNECT_TIMEOUT`, src/netbase.h at
# bitcoin/bitcoin@ca7162cde5), not the old poll loop's ten-passes-at-0.1s
# budget this constant carried until ISS 681: that budget was never
# itself checked against Core, and it was too tight for what
# `loop.sock_connect` needs on Windows' Proactor loop to notice a
# refused loopback connect -- measured on this tree's own
# instrumentation, in btclib-org/btclib-node run 33271519023, at
# elapsed=2.017810 for a `ConnectionRefusedError` the run's own kernel
# raised. `loop.sock_connect` does not need the two magic numbers a poll
# needs, only this one: a real timeout wrapped around a wait that is
# otherwise event-driven.
_DIAL_TIMEOUT = 5.0
[docs]
async def dial(address: NetworkAddressV2) -> socket.socket | None:
"""Return a socket connected to the peer, or nothing if it never came up.
`dial` and not `connect`, which is what `P2pManager` calls the whole
of making a connection out of one: this is the socket alone.
`loop.sock_connect` is the kernel's own answer rather than a guess at
it: a refusal is `SO_ERROR` on the socket, read the moment the OS
notifies the loop's writer callback, not inferred after a fixed
number of `getpeername` polls that cannot tell a refusal from a peer
that is merely slow. And where `connect` completes without ever
raising `BlockingIOError` -- a local peer most often -- `sock_connect`
returns at once instead of an `except` arm that never runs.
No separate check for a host with no route to the family being
dialled: `_DIAL_TIMEOUT` already bounds every attempt, and an
unreachable family fails the same `sock_connect` a slow or refusing
peer does, landing on the same `None` `P2pManager` already treats as
"try someone else". Bitcoin Core's own default (`ReachableNets`,
src/netbase.h at 58a7869f86: "Everything is reachable by default")
is the same bet -- reachability is what a dial's outcome says it is,
not a property guessed at beforehand -- so there is nothing here for
a heavier check to buy.
"""
if address.network_id not in _IP_NETWORKS:
raise UnsupportedAddressTypeError
if address.network_id == BIP155Network.IPV4:
family = socket.AF_INET
host = str(IPv4Address(address.address))
peer: tuple[str, int] | tuple[str, int, int, int] = (host, address.port)
else:
family = socket.AF_INET6
host = str(IPv6Address(address.address))
# ISS 682: a bare 2-tuple is what a POSIX `socket.connect()`
# accepts for an IPv6 peer, defaulting flowinfo and scope id to
# 0. Windows' Proactor loop hands this straight to `ConnectEx`
# instead, and CPython's `Modules/overlapped.c`
# (`parse_address`, read at python/cpython@v3.14.0) dispatches
# on the tuple's length alone rather than on the socket's own
# family: a 2-tuple is always parsed as `AF_INET`, so
# `WSAStringToAddressW` is asked to read "::1" as an IPv4
# dotted quad and answers `WSAEINVAL` synchronously, before
# `ConnectEx` is ever reached -- confirmed from
# btclib-org/btclib-node run 33270966438's own instrumentation:
# `family=23 ... elapsed=0.000099 exc=OSError(22, 'An invalid
# argument was supplied', None, 10022, None)`. The four-tuple
# form names the family explicitly and is accepted on every
# platform this node runs on, POSIX included.
peer = (host, address.port, 0, 0)
client = socket.socket(family, socket.SOCK_STREAM)
client.settimeout(0)
loop = asyncio.get_running_loop()
try:
await asyncio.wait_for(loop.sock_connect(client, peer), _DIAL_TIMEOUT)
except OSError, TimeoutError:
client.close()
return None
except asyncio.CancelledError:
# `manage_connections` cancelled with a dial in flight, which is
# what `P2pManager.stop`'s own drain does to it. This socket is
# this call's own until it is handed back, and the caller that
# never receives it has nothing to close: without this it goes
# out with the frame the cancellation unwinds, and is reported
# against whichever test the collector reaches it in
# (btclib-org/btclib-node#312).
client.close()
raise
return client
# Two record kinds share the one store `PeerDB` opens, so `init_from_db`
# below walks it whole and dispatches on the prefix rather than stopping
# at the first key without one -- `src/btclib_node/db.py`'s own docstring
# names that shape as `BlockDB`'s, next to the other one, `BlockIndex`'s,
# that a store of one record kind can use instead.
_KNOWN = b"known-"
_ANSWERED = b"answered-"
# The bound both tables are kept under: an address a peer gossiped, and
# an address this node has itself confirmed reachable. The cap is on
# distinct endpoints, not on handshakes: `add_active_address` settles a
# repeat handshake with the same endpoint onto the one row already held
# for it (#270), the way `add_addresses`'s own `by_endpoint` already
# does for `self.addresses`.
_MAX_ADDRESSES = 10000
[docs]
def endpoint_key(address: NetworkAddressV2) -> bytes:
"""Return the octets a persisted address is keyed on.
The network id, the address and the port -- what names an endpoint
on the wire -- and not `timestamp` or `services`: those are this
node's own opinion of the endpoint, not part of what it is, so two
records differing only in them settle on the one row written last
rather than growing the table an entry per gossip or per reconnect.
"""
endpoint = replace(address, timestamp=0, services=ServiceFlags.NODE_NONE)
return endpoint.serialize(check_validity=False)
[docs]
class PeerDB:
"""The table of addresses this node knows of, gossiped and self-confirmed.
`addresses` is every address heard about; `active_addresses` is the
subset this node has itself dialled and heard back from recently.
Each is behind its own lock, taken separately and never nested --
the comment beside each lock's own field says which thread reaches
it and why sharing the other lock was declined.
"""
def __init__(self, chain: Chain, data_dir: Path | None) -> None:
"""Load the durable tables, then decide whether DNS is still needed."""
self.chain = chain
self.data_dir = data_dir
self.addresses: set[NetworkAddressV2] = set()
# A lock of its own, not `_active_lock` below: `add_addresses`
# reaches this set from both threads too (#298) -- gossip
# through `callbacks.addr`/`addrv2` on `Node`'s, DNS seed
# answers through `get_addr_from_dns` on `P2pManager`'s, and
# `random_address`'s own dialable-address comprehension on
# `P2pManager`'s as well, racing against gossip on `Node`'s.
# Unprotected, that last pairing is not only the lost-update or
# wrong-row risk `_active_lock` guards against: iterating a
# `set` while another thread mutates it is `RuntimeError: Set
# changed size during iteration` in CPython, a crash rather than
# a silent corruption. Sharing `_active_lock` instead was
# measured and declined: nothing here ever needs the two tables
# updated as one atomic step, and `add_addresses`'s own durable
# write batch is measurably slower than `add_active_address`'s
# single row -- sharing would let it hold up a handshake for no
# invariant this table's own lock does not already give it.
self._addresses_lock = threading.Lock()
self.active_addresses: list[NetworkAddressV2] = []
# endpoint bytes -> its position in `active_addresses`, so
# `add_active_address` can find a repeat endpoint's row in O(1)
# rather than by scanning the list it is called once per
# handshake against (#270). Rebuilt rather than kept in step
# wherever something else reshapes the list instead --
# `init_from_db`'s bulk load and `get_active_addresses`'s prune,
# both already O(n) over it.
self._active_index: dict[bytes, int] = {}
# `add_active_address` reads this index and then writes into
# `active_addresses` at the position it found -- two statements,
# not one -- and `get_active_addresses` reassigns the list and
# then rebuilds the index against it -- likewise two. The first
# runs on `Node`'s own thread, off `callbacks.verack`; the
# second runs on `P2pManager`'s, off `manage_connections`, which
# calls it every few minutes regardless of what else that loop
# is doing (#71). Interleaved without a lock, a position read
# before a prune can be written after it, into a list the prune
# already reshaped: one endpoint's row silently holding another
# endpoint's data, or an `IndexError`. `KeyValueStore` has its
# own lock for the store; this one is for these two in-memory
# structures alone, and is not the same lock.
self._active_lock = threading.Lock()
# What `callbacks.getaddr` last answered with, and until when it
# is still good for: a fresh `secrets.SystemRandom().sample` per
# connection would let two peers connecting close together
# compare answers and infer what changed between them, which
# "once per connection" alone does not stop -- a new connection
# still draws fresh. `0.0` starts already expired, so the first
# call computes a sample rather than serving an empty one.
# btclib-org/btclib-node#71
self.addr_sample: list[NetworkAddressV2] = []
self.addr_sample_expiration = 0.0
# `None` is a table kept in memory only, which is what every
# test here wants and what `data_dir` was before this: assigned
# and never opened. `Node` always passes an actual directory
# (`Config.data_dir` has no `None` of its own), so this is the
# one place that distinction is made.
self.db = KeyValueStore(data_dir / "peers") if data_dir is not None else None
self.init_from_db()
# DNS is asked only where the durable table came back with
# nothing this node has itself confirmed working recently:
# `get_active_addresses` is what "recently" already means, and
# `can_connect` is what catches a table `add_addresses` filled
# with tor, i2p or an ipv6-only answer from a seed -- #89, where
# a nonempty table was exactly the case DNS was skipped for and
# none of it was dialable.
self.ask_dns_nodes = not any(
can_connect(address) for address in self.get_active_addresses()
)
[docs]
def init_from_db(self) -> None:
"""Load every stored address into `addresses` or `active_addresses`.
One store keyed by two prefixes (the comment on `_KNOWN` and
`_ANSWERED` above argues why), so this walks it whole and
dispatches on the prefix rather than stopping at the first key
without one.
"""
if self.db is None:
return
for key, value in self.db:
if key.startswith(_KNOWN):
self.addresses.add(NetworkAddressV2.parse(value, check_validity=False))
elif key.startswith(_ANSWERED):
self.active_addresses.append(
NetworkAddressV2.parse(value, check_validity=False)
)
self._reindex_active()
def _reindex_active(self) -> None:
"""Rebuild the endpoint index over the current `active_addresses`.
O(n), the same order `get_active_addresses`'s own prune already
walks the list at -- called from there and from `init_from_db`,
the two places that reshape the list itself rather than through
`add_active_address`.
"""
self._active_index = {
endpoint_key(address): position
for position, address in enumerate(self.active_addresses)
}
@contextmanager
def _write_batch(self) -> Iterator[KeyValueStore | None]:
"""Yield a batch to write into, or `None` where there is nothing to.
One shape either way, so a caller writes `if wb is not None`
around the puts it makes and nothing around the batch itself --
the alternative, a call site branching on `self.db` before ever
reaching a loop, is what this exists to not be.
"""
if self.db is None:
yield None
return
with self.db.write_batch() as wb:
yield wb
[docs]
def close(self) -> None:
"""Close the durable store, if this table has one."""
if self.db is not None:
self.db.close()
[docs]
async def get_addr_from_dns(self) -> None:
"""Resolve every chain DNS seed and feed the answers to `add_addresses`.
A no-op unless `ask_dns_nodes` said, at construction time, that
the durable table came back with nothing dialable.
"""
if not self.ask_dns_nodes:
return
chain = self.chain
loop = asyncio.get_running_loop()
# what a seed answers with, deduplicated: seeds overlap, and one
# of them answers with the same host over several records.
endpoints: set[tuple[str, int]] = set()
for dns_server in chain.addresses:
try:
answers = await loop.getaddrinfo(dns_server, chain.port)
except socket.gaierror:
continue
# (family, type, proto, canonname, sockaddr), and the
# sockaddr is the only part a peer table wants. It opens
# with the host and the port -- two fields for AF_INET,
# four for AF_INET6, whose flow info and scope id say
# nothing a BIP155 record holds. The stub also admits
# AF_PACKET's (protocol, address) pair, which resolving an
# internet host and a port cannot answer with, so the cast
# is what that fact is written as rather than a check no
# test could reach.
for *_, sockaddr in answers:
endpoints.add(cast("tuple[str, int]", sockaddr[:2]))
# through add_addresses, and not a bare add to the set: a seed
# is gossip like a peer's is, and belongs in the durable table
# the same way, so a later restart has it without asking again
self.add_addresses(peer_address(ip, port) for ip, port in endpoints)
@property
def is_empty(self) -> bool:
"""Whether `addresses` holds nothing at all, read without a lock."""
# Unlocked on purpose: `len` on a set is one step, not a walk of
# it, so there is nothing here for another thread's `add`/
# `discard` to catch mid-stride -- the answer is at worst one
# mutation stale, the same imprecision `manage_connections`
# already reads this property through (a table this answers
# empty for can gain an entry the instant after, dialable or
# not, and nothing here promised otherwise).
return not len(self.addresses)
[docs]
def random_address(self) -> NetworkAddressV2 | None:
"""Return a random dialable address, or `None` if there is none.
Preferred from `get_active_addresses`'s own dialable subset;
falls back to `addresses` whole, locked, only if that is empty.
"""
# Preferred: an address this node has itself dialled and heard
# back from recently, over one merely gossiped -- #123, so that
# a run draws on what it already knows works rather than on the
# whole table uniformly, even before a restart ever reads any
# of it back.
preferred = [addr for addr in self.get_active_addresses() if can_connect(addr)]
if preferred:
return secrets.choice(preferred)
# Drawn from the addresses that can be dialled, rather than from
# the whole table with a retry on the ones that cannot: a table
# holding none of them -- a seed answering with AAAA records
# alone is enough, and `add_addresses` takes whatever tor, i2p
# and cjdns a peer sends -- made that retry a loop with no exit,
# in the caller's event loop. Nothing to dial is an answer, and
# `None` is it.
# Locked, unlike `is_empty` above: this walks the set rather
# than asking its length, and add_addresses (#298) reaches it
# from Node's own thread while this runs on P2pManager's --
# unprotected, that is CPython's `RuntimeError: Set changed
# size during iteration`, not merely a stale answer.
with self._addresses_lock:
dialable = [address for address in self.addresses if can_connect(address)]
if not dialable:
return None
return secrets.choice(dialable)
[docs]
def add_addresses(self, addresses: Iterable[NetworkAddressV2]) -> None:
"""Merge `addresses` into `self.addresses`, checked and deduplicated.
BIP155's embedded-IPv6 records are dropped; every other address
settles onto its own `endpoint_key` row, up to `_MAX_ADDRESSES`
distinct endpoints, past which a genuinely new one is dropped
too. Locked with `_addresses_lock`.
"""
# a peer's word for when it last saw an address is not evidence,
# and keeping it would make the one address several entries
with self._addresses_lock, self._write_batch() as wb:
# `endpoint_key` is what the durable row is already keyed on --
# network id, address and port, not `services` -- so a
# second gossip for the one endpoint overwrites the row on
# disk. This index is what makes `self.addresses` settle on
# the endpoint the same way instead of holding one member
# per `services` value ever seen for it (#247).
by_endpoint = {endpoint_key(known): known for known in self.addresses}
for address in addresses:
# BIP155's ignore rule: an IPV6 record that is really an
# IPv4 or a (long-retired) TORv2 address wearing another
# network's sixteen octets is not a second peer, and
# keeping it under network id 2 is what used to gossip
# it back as IPv4 -- the same host, twice in the table
# (#151). Checked before the durable write too, so a
# dropped record is dropped everywhere, not merely kept
# out of the in-memory set.
if _is_embedded_ipv6(address):
continue
known = replace(address, timestamp=0)
key = endpoint_key(known)
existing = by_endpoint.get(key)
# the cap is on distinct endpoints, so updating one
# already held does not spend it -- only a genuinely new
# endpoint can run the table out of room
if existing is None and len(self.addresses) >= _MAX_ADDRESSES:
break
if existing is not None:
self.addresses.discard(existing)
self.addresses.add(known)
by_endpoint[key] = known
if wb is not None:
value = known.serialize(check_validity=False)
wb.put(_KNOWN + key, value)
[docs]
def get_active_addresses(self) -> list[NetworkAddressV2]:
"""Return `active_addresses`, pruned of every entry older than 3 hours.
A pruned entry's durable `answered-` row is deleted too. Locked
with `_active_lock`.
"""
now = time.time()
with self._active_lock:
# active if seen within the last three hours; an entry that
# ages out here loses its `answered-` row too, so the
# durable store stays bounded by what is still active rather
# than by every endpoint this node has ever dialled and
# heard back from over its whole lifetime (#253)
active: list[NetworkAddressV2] = []
for addr in self.active_addresses:
if now - addr.timestamp < 3600 * 3:
active.append(addr)
elif self.db is not None:
self.db.delete(_ANSWERED + endpoint_key(addr))
self.active_addresses = active
self._reindex_active()
return self.active_addresses
[docs]
def add_active_address(self, addr: NetworkAddressV2) -> None:
"""Record `addr` as dialled and answered, just now.
A repeat handshake with an already-held endpoint settles onto
its one row rather than growing the table. Locked with
`_active_lock`.
"""
# a whole second: the field is four octets on the wire
answered = replace(addr, timestamp=int(time.time()))
key = endpoint_key(answered)
with self._active_lock:
position = self._active_index.get(key)
if position is not None:
# a repeat handshake with an endpoint already held:
# settle onto its one row rather than growing the table
# an entry per reconnect (#270), matching
# `add_addresses`'s own `by_endpoint`.
self.active_addresses[position] = answered
else:
# the cap is on distinct endpoints, so updating one
# already held (above) does not spend it -- only a
# genuinely new one can run the table out of room,
# `add_addresses`'s own cap check reads the same way.
if len(self.active_addresses) >= _MAX_ADDRESSES:
return
self._active_index[key] = len(self.active_addresses)
self.active_addresses.append(answered)
if self.db is not None:
self.db.put(_ANSWERED + key, answered.serialize(check_validity=False))