mirror of
https://github.com/waku-org/nwaku.git
synced 2025-01-14 17:04:53 +00:00
deploy: e85b5cbae1875c3c036f7cf12453f88031ffa592
This commit is contained in:
LNSD
parent
ccd08c16af
commit
68aeda48c3
@ -15,8 +15,6 @@ import
|
||||
eth/keys
|
||||
import
|
||||
../../waku/v2/node/waku_node,
|
||||
../../waku/v2/node/message_store/queue_store,
|
||||
../../waku/v2/protocol/waku_store,
|
||||
../../waku/v2/protocol/waku_swap/waku_swap,
|
||||
../../waku/v2/utils/peers,
|
||||
../test_helpers,
|
||||
@ -50,106 +48,3 @@ procSuite "Waku SWAP Accounting":
|
||||
check:
|
||||
decodedCheque.isErr == false
|
||||
decodedCheque.get() == cheque
|
||||
|
||||
# TODO: To do this reliably we need access to contract node
|
||||
# With current logic state isn't updated because of bad cheque
|
||||
# Consider moving this test to e2e test, and/or move swap module to be on by default
|
||||
asyncTest "Update accounting state after store operations":
|
||||
## Setup
|
||||
let
|
||||
serverKey = crypto.PrivateKey.random(Secp256k1, rng[])[]
|
||||
server = WakuNode.new(serverKey, ValidIpAddress.init("0.0.0.0"), Port(60102))
|
||||
clientKey = crypto.PrivateKey.random(Secp256k1, rng[])[]
|
||||
client = WakuNode.new(clientKey, ValidIpAddress.init("0.0.0.0"), Port(60100))
|
||||
|
||||
await allFutures(client.start(), server.start())
|
||||
|
||||
await server.mountSwap()
|
||||
await server.mountStore(store=StoreQueueRef.new())
|
||||
await client.mountSwap()
|
||||
await client.mountStore()
|
||||
client.mountStoreClient()
|
||||
|
||||
client.wakuSwap.setPeer(server.peerInfo.toRemotePeerInfo())
|
||||
server.wakuSwap.setPeer(client.peerInfo.toRemotePeerInfo())
|
||||
|
||||
client.setStorePeer(server.peerInfo.toRemotePeerInfo())
|
||||
server.setStorePeer(client.peerInfo.toRemotePeerInfo())
|
||||
|
||||
## Given
|
||||
let message = fakeWakuMessage()
|
||||
require server.wakuStore.store.put(DefaultPubsubTopic, message).isOk()
|
||||
|
||||
let serverPeer = server.peerInfo.toRemotePeerInfo()
|
||||
let req = HistoryQuery(contentTopics: @[DefaultContentTopic])
|
||||
|
||||
## When
|
||||
let queryRes = await client.query(req, peer=serverPeer)
|
||||
|
||||
## Then
|
||||
check queryRes.isOk()
|
||||
|
||||
let response = queryRes.get()
|
||||
check:
|
||||
response.messages == @[message]
|
||||
|
||||
check:
|
||||
client.wakuSwap.accounting[server.peerInfo.peerId] == 1
|
||||
server.wakuSwap.accounting[client.peerInfo.peerId] == -1
|
||||
|
||||
## Cleanup
|
||||
await allFutures(client.stop(), server.stop())
|
||||
|
||||
|
||||
# This test will only Be checked if in Mock mode
|
||||
# TODO: Add cheque here
|
||||
asyncTest "Update accounting state after sending cheque":
|
||||
## Setup
|
||||
let
|
||||
serverKey = crypto.PrivateKey.random(Secp256k1, rng[])[]
|
||||
server = WakuNode.new(serverKey, ValidIpAddress.init("0.0.0.0"), Port(60202))
|
||||
clientKey = crypto.PrivateKey.random(Secp256k1, rng[])[]
|
||||
client = WakuNode.new(clientKey, ValidIpAddress.init("0.0.0.0"), Port(60200))
|
||||
|
||||
# Define the waku swap Config for this test
|
||||
let swapConfig = SwapConfig(mode: SwapMode.Mock, paymentThreshold: 1, disconnectThreshold: -1)
|
||||
|
||||
# Start nodes and mount protocols
|
||||
await allFutures(client.start(), server.start())
|
||||
await server.mountSwap(swapConfig)
|
||||
await server.mountStore(store=StoreQueueRef.new())
|
||||
await client.mountSwap(swapConfig)
|
||||
await client.mountStore()
|
||||
client.mountStoreClient()
|
||||
|
||||
client.wakuSwap.setPeer(server.peerInfo.toRemotePeerInfo())
|
||||
server.wakuSwap.setPeer(client.peerInfo.toRemotePeerInfo())
|
||||
|
||||
client.setStorePeer(server.peerInfo.toRemotePeerInfo())
|
||||
server.setStorePeer(client.peerInfo.toRemotePeerInfo())
|
||||
|
||||
## Given
|
||||
let message = fakeWakuMessage()
|
||||
require server.wakuStore.store.put(DefaultPubsubTopic, message).isOk()
|
||||
|
||||
let serverPeer = server.peerInfo.toRemotePeerInfo()
|
||||
let req = HistoryQuery(contentTopics: @[DefaultContentTopic])
|
||||
|
||||
## When
|
||||
# TODO: Handshakes - for now we assume implicit, e2e still works for PoC
|
||||
let res1 = await client.query(req, peer=serverPeer)
|
||||
let res2 = await client.query(req, peer=serverPeer)
|
||||
|
||||
require:
|
||||
res1.isOk()
|
||||
res2.isOk()
|
||||
|
||||
## Then
|
||||
check:
|
||||
# Accounting table updated with credit and debit, respectively
|
||||
# After sending a cheque the balance is partially adjusted
|
||||
client.wakuSwap.accounting[server.peerInfo.peerId] == 1
|
||||
server.wakuSwap.accounting[client.peerInfo.peerId] == -1
|
||||
|
||||
## Cleanup
|
||||
await allFutures(client.stop(), server.stop())
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
# libtool - Provide generalized library-building support services.
|
||||
# Generated automatically by config.status (libbacktrace) version-unused
|
||||
# Libtool was configured on host fv-az41-370:
|
||||
# Libtool was configured on host fv-az508-423:
|
||||
# NOTE: Changes made to this file will be lost: look at ltmain.sh.
|
||||
#
|
||||
# Copyright (C) 1996, 1997, 1998, 1999, 2000, 2001, 2003, 2004, 2005,
|
||||
|
||||
@ -572,16 +572,12 @@ proc startMessageRetentionPolicyPeriodicTask*(node: WakuNode, interval: Duration
|
||||
discard setTimer(Moment.fromNow(interval), executeRetentionPolicy)
|
||||
|
||||
proc mountStore*(node: WakuNode, store: MessageStore = nil, retentionPolicy=none(MessageRetentionPolicy) ) {.async, raises: [Defect, LPError].} =
|
||||
if node.wakuSwap.isNil():
|
||||
info "mounting waku store protocol (no waku swap)"
|
||||
else:
|
||||
info "mounting waku store protocol with waku swap support"
|
||||
info "mounting waku store protocol"
|
||||
|
||||
node.wakuStore = WakuStore.new(
|
||||
node.peerManager,
|
||||
node.rng,
|
||||
store,
|
||||
wakuSwap=node.wakuSwap,
|
||||
retentionPolicy=retentionPolicy
|
||||
)
|
||||
|
||||
@ -608,10 +604,6 @@ proc query*(node: WakuNode, query: HistoryQuery, peer: RemotePeerInfo): Future[W
|
||||
|
||||
let response = queryRes.get()
|
||||
|
||||
if not node.wakuSwap.isNil():
|
||||
# Perform accounting operation
|
||||
node.wakuSwap.debit(peer.peerId, response.messages.len)
|
||||
|
||||
return ok(response)
|
||||
|
||||
|
||||
|
||||
@ -15,7 +15,6 @@ import
|
||||
../../utils/requests,
|
||||
../../utils/time,
|
||||
../waku_message,
|
||||
../waku_swap/waku_swap,
|
||||
./protocol_metrics,
|
||||
./common,
|
||||
./rpc,
|
||||
@ -35,7 +34,6 @@ type WakuStoreClient* = ref object
|
||||
peerManager: PeerManager
|
||||
rng: ref rand.HmacDrbgContext
|
||||
store: MessageStore
|
||||
wakuSwap: WakuSwap
|
||||
|
||||
proc new*(T: type WakuStoreClient,
|
||||
peerManager: PeerManager,
|
||||
|
||||
@ -22,7 +22,6 @@ import
|
||||
../../node/peer_manager/peer_manager,
|
||||
../../utils/time,
|
||||
../waku_message,
|
||||
../waku_swap/waku_swap,
|
||||
./common,
|
||||
./rpc,
|
||||
./rpc_codec,
|
||||
@ -43,7 +42,6 @@ type
|
||||
peerManager*: PeerManager
|
||||
rng*: ref rand.HmacDrbgContext
|
||||
store*: MessageStore
|
||||
wakuSwap*: WakuSwap
|
||||
retentionPolicy: Option[MessageRetentionPolicy]
|
||||
|
||||
|
||||
@ -207,7 +205,7 @@ proc findMessages(w: WakuStore, query: HistoryQuery): HistoryResult {.gcsafe.} =
|
||||
|
||||
## Protocol
|
||||
|
||||
proc initProtocolHandler*(ws: WakuStore) =
|
||||
proc initProtocolHandler(ws: WakuStore) =
|
||||
|
||||
proc handler(conn: Connection, proto: string) {.async.} =
|
||||
let buf = await conn.readLp(MaxRpcSize.int)
|
||||
@ -259,16 +257,6 @@ proc initProtocolHandler*(ws: WakuStore) =
|
||||
|
||||
let resp = respRes.toRPC()
|
||||
|
||||
if not ws.wakuSwap.isNil():
|
||||
info "handle store swap", peerId=conn.peerId, requestId=reqRpc.requestId, text=ws.wakuSwap.text
|
||||
|
||||
# Perform accounting operation
|
||||
# TODO: Do accounting here, response is HistoryResponseRPC. How do we get node or swap context?
|
||||
let peerId = conn.peerId
|
||||
let messages = resp.messages
|
||||
ws.wakuSwap.credit(peerId, messages.len)
|
||||
|
||||
|
||||
info "sending history response", peerId=conn.peerId, requestId=reqRpc.requestId, messages=resp.messages.len
|
||||
|
||||
let rpc = HistoryRPC(requestId: reqRpc.requestId, response: some(resp))
|
||||
@ -281,13 +269,11 @@ proc new*(T: type WakuStore,
|
||||
peerManager: PeerManager,
|
||||
rng: ref rand.HmacDrbgContext,
|
||||
store: MessageStore,
|
||||
wakuSwap: WakuSwap = nil,
|
||||
retentionPolicy=none(MessageRetentionPolicy)): T =
|
||||
let ws = WakuStore(
|
||||
rng: rng,
|
||||
peerManager: peerManager,
|
||||
store: store,
|
||||
wakuSwap: wakuSwap,
|
||||
retentionPolicy: retentionPolicy
|
||||
)
|
||||
ws.initProtocolHandler()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user