fix: modify unsubscribe cleanup routine and tests (#84)

* fix: modify unsubscribe cleanup routine

Ignore exceptions (other than CancelledError) if uninstallation of the filter fails. If it's the last step in the subscription cleanup, then filter changes for this filter will no longer be polled so if the filter continues to live on in geth for whatever reason, then it doesn't matter.

This includes a number of fixes:
- `CancelledError` is now caught inside of `getChanges`. This was causing conditions during `subscriptions.close`, where the `CancelledError` would get consumed by the `except CatchableError`, if there was an ongoing `poll` happening at the time of close.
- After creating a new filter inside of `getChanges`, the new filter is polled for changes before returning.
- `getChanges` also does not swallow `CatchableError` by returning an empty array, and instead re-raises the error if it is not `filter not found`.
- The tests were simplified by accessing the private fields of `PollingSubscriptions`. That way, there wasn't a race condition for the `newFilterId` counter inside of the mock.
- The `MockRpcHttpServer` was simplified by keeping track of the active filters only, and invalidation simply removes the filter. The tests then only needed to rely on the fact that the filter id changed in the mapping.
- Because of the above changes, we no longer needed to sleep inside of the tests, so the sleeps were removed, and the polling interval could be changed to 1ms, which not only makes the tests faster, but would further highlight any race conditions if present.

* docs: rpc custom port documentation

---------

Co-authored-by: Adam Uhlíř <adam@uhlir.dev>

.gitignore

@@ -5,3 +5,4 @@ nimble.develop
 nimble.paths
 .idea
 .nimble
+.envrc

Readme.md

@@ -202,6 +202,9 @@ $ npm ci
 $ npm start
 ```
 
+If you need to use different port for the RPC node, then you can start with `npm start -- --port 1111` and
+then run the tests with `ETHERS_TEST_PROVIDER=1111 nimble test`.
+
 Thanks
 ------
 

ethers/providers/jsonrpc/subscriptions.nim

@@ -17,6 +17,7 @@ type
     methodHandlers: Table[string, MethodHandler]
   MethodHandler* = proc (j: JsonNode) {.gcsafe, raises: [].}
   SubscriptionCallback = proc(id, arguments: JsonNode) {.gcsafe, raises:[].}
+  SubscriptionError* = object of EthersError
 
 {.push raises:[].}
 
@@ -130,7 +131,7 @@ method unsubscribe*(subscriptions: WebSocketSubscriptions,
 # Polling
 
 type
-  PollingSubscriptions = ref object of JsonRpcSubscriptions
+  PollingSubscriptions* = ref object of JsonRpcSubscriptions
     polling: Future[void]
 
     # We need to keep around the filters that are used to create log filters on the RPC node
@@ -151,14 +152,24 @@ proc new*(_: type JsonRpcSubscriptions,
   proc getChanges(originalId: JsonNode): Future[JsonNode] {.async.} =
     try:
       let mappedId = subscriptions.subscriptionMapping[originalId]
-      return await subscriptions.client.eth_getFilterChanges(mappedId)
+      let changes = await subscriptions.client.eth_getFilterChanges(mappedId)
+      if changes.kind == JNull:
+        return newJArray()
+      elif changes.kind != JArray:
+        raise newException(SubscriptionError,
+          "HTTP polling: unexpected value returned from eth_getFilterChanges." &
+          " Expected: JArray, got: " & $changes.kind)
+      return changes
+    except CancelledError as e:
+      raise e
     except CatchableError as e:
       if "filter not found" in e.msg:
         let filter = subscriptions.filters[originalId]
         let newId = await subscriptions.client.eth_newFilter(filter)
         subscriptions.subscriptionMapping[originalId] = newId
-
+        return await getChanges(originalId)
-      return newJArray()
+      else:
+        raise e
 
   proc poll(id: JsonNode) {.async.} =
     for change in await getChanges(id):
@@ -218,7 +229,17 @@ method subscribeLogs(subscriptions: PollingSubscriptions,
 method unsubscribe*(subscriptions: PollingSubscriptions,
                    id: JsonNode)
                   {.async.} =
-  discard await subscriptions.client.eth_uninstallFilter(subscriptions.subscriptionMapping[id])
   subscriptions.filters.del(id)
   subscriptions.callbacks.del(id)
+  let sub = subscriptions.subscriptionMapping[id]
   subscriptions.subscriptionMapping.del(id)
+  try:
+    discard await subscriptions.client.eth_uninstallFilter(sub)
+  except CancelledError as e:
+    raise e
+  except CatchableError:
+    # Ignore if uninstallation of the filter fails. If it's the last step in our
+    # cleanup, then filter changes for this filter will no longer be polled so
+    # if the filter continues to live on in geth for whatever reason then it
+    # doesn't matter.
+    discard

testmodule/providers/jsonrpc/rpc_mock.nim

@@ -3,40 +3,39 @@ import ../../../ethers/provider
 import ../../../ethers/providers/jsonrpc/conversions
 
 import std/tables
+import std/sequtils
 import pkg/stew/byteutils
 import pkg/json_rpc/rpcserver except `%`, `%*`
 import pkg/json_rpc/errors
 
-
 type MockRpcHttpServer* = ref object
-  filters*: Table[string, bool]
+  filters*: seq[string]
-  newFilterCounter*: int
   srv: RpcHttpServer
 
 proc new*(_: type MockRpcHttpServer): MockRpcHttpServer =
-  MockRpcHttpServer(filters: initTable[string, bool](), newFilterCounter: 0, srv: newRpcHttpServer(["127.0.0.1:0"]))
+  let srv = newRpcHttpServer(["127.0.0.1:0"])
+  MockRpcHttpServer(filters: @[], srv: srv)
 
-proc invalidateFilter*(server: MockRpcHttpServer, id: string) =
+proc invalidateFilter*(server: MockRpcHttpServer, jsonId: JsonNode) =
-  server.filters[id] = false
+  server.filters.keepItIf it != jsonId.getStr
 
 proc start*(server: MockRpcHttpServer) =
   server.srv.router.rpc("eth_newFilter") do(filter: EventFilter) -> string:
     let filterId = "0x" & (array[16, byte].example).toHex
-    server.filters[filterId] = true
+    server.filters.add filterId
-    server.newFilterCounter += 1
     return filterId
 
   server.srv.router.rpc("eth_getFilterChanges") do(id: string) -> seq[string]:
-    if(not hasKey(server.filters, id) or not server.filters[id]):
+    if id notin server.filters:
       raise (ref ApplicationError)(code: -32000, msg: "filter not found")
 
     return @[]
 
   server.srv.router.rpc("eth_uninstallFilter") do(id: string) -> bool:
-    if(not hasKey(server.filters, id)):
+    if id notin server.filters:
       raise (ref ApplicationError)(code: -32000, msg: "filter not found")
 
-    server.filters.del(id)
+    server.invalidateFilter(%id)
     return true
 
   server.srv.start()
@@ -45,6 +44,5 @@ proc stop*(server: MockRpcHttpServer) {.async.} =
   await server.srv.stop()
   await server.srv.closeWait()
 
-
 proc localAddress*(server: MockRpcHttpServer): seq[TransportAddress] =
   return server.srv.localAddress()

testmodule/providers/jsonrpc/testJsonRpcProvider.nim

@@ -1,3 +1,4 @@
+import std/os
 import pkg/asynctest
 import pkg/chronos
 import pkg/ethers
@@ -6,7 +7,8 @@ import pkg/stew/byteutils
 import ../../examples
 import ../../miner
 
-for url in ["ws://localhost:8545", "http://localhost:8545"]:
+let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
+for url in ["ws://" & providerUrl, "http://"  & providerUrl]:
 
   suite "JsonRpcProvider (" & url & ")":
 

testmodule/providers/jsonrpc/testJsonRpcSigner.nim

@@ -1,3 +1,4 @@
+import std/os
 import pkg/asynctest
 import pkg/ethers
 import pkg/stew/byteutils
@@ -7,9 +8,10 @@ suite "JsonRpcSigner":
 
   var provider: JsonRpcProvider
   var accounts: seq[Address]
+  let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
 
   setup:
-    provider = JsonRpcProvider.new()
+    provider = JsonRpcProvider.new("http://" & providerUrl)
     accounts = await provider.listAccounts()
 
   teardown:

testmodule/providers/jsonrpc/testJsonRpcSubscriptions.nim

@@ -1,5 +1,7 @@
 import std/json
+import std/os
 import std/sequtils
+import std/importutils
 import pkg/asynctest
 import pkg/serde
 import pkg/json_rpc/rpcclient
@@ -68,7 +70,7 @@ suite "Web socket subscriptions":
 
   setup:
     client = newRpcWebSocketClient()
-    await client.connect("ws://localhost:8545")
+    await client.connect("ws://"  & getEnv("ETHERS_TEST_PROVIDER", "localhost:8545"))
     subscriptions = JsonRpcSubscriptions.new(client)
     subscriptions.start()
 
@@ -85,7 +87,7 @@ suite "HTTP polling subscriptions":
 
   setup:
     client = newRpcHttpClient()
-    await client.connect("http://localhost:8545")
+    await client.connect("http://" & getEnv("ETHERS_TEST_PROVIDER", "localhost:8545"))
     subscriptions = JsonRpcSubscriptions.new(client,
                                              pollingInterval = 100.millis)
     subscriptions.start()
@@ -98,10 +100,12 @@ suite "HTTP polling subscriptions":
 
 suite "HTTP polling subscriptions - filter not found":
 
-  var subscriptions: JsonRpcSubscriptions
+  var subscriptions: PollingSubscriptions
   var client: RpcHttpClient
   var mockServer: MockRpcHttpServer
 
+  privateAccess(PollingSubscriptions)
+
   setup:
     mockServer = MockRpcHttpServer.new()
     mockServer.start()
@@ -109,8 +113,10 @@ suite "HTTP polling subscriptions - filter not found":
     client = newRpcHttpClient()
     await client.connect("http://" & $mockServer.localAddress()[0])
 
-    subscriptions = JsonRpcSubscriptions.new(client,
+    subscriptions = PollingSubscriptions(
-                                             pollingInterval = 15.millis)
+                      JsonRpcSubscriptions.new(
+                        client,
+                        pollingInterval = 1.millis))
     subscriptions.start()
 
   teardown:
@@ -122,35 +128,28 @@ suite "HTTP polling subscriptions - filter not found":
     let filter = EventFilter(address: Address.example, topics: @[array[32, byte].example])
     let emptyHandler = proc(log: Log) = discard
 
-    check mockServer.newFilterCounter == 0
+    check subscriptions.filters.len == 0
-    let jsonId = await subscriptions.subscribeLogs(filter, emptyHandler)
+    check subscriptions.subscriptionMapping.len == 0
-    let id = string.fromJson(jsonId).tryGet
+
-    check mockServer.newFilterCounter == 1
+    let id = await subscriptions.subscribeLogs(filter, emptyHandler)
+
+    check subscriptions.filters[id] == filter
+    check subscriptions.subscriptionMapping[id] == id
+    check subscriptions.filters.len == 1
+    check subscriptions.subscriptionMapping.len == 1
 
-    await sleepAsync(50.millis)
     mockServer.invalidateFilter(id)
-    await sleepAsync(50.millis)
+
-    check mockServer.newFilterCounter == 2
+    check eventually subscriptions.subscriptionMapping[id] != id
 
   test "recreated filter can be still unsubscribed using the original id":
     let filter = EventFilter(address: Address.example, topics: @[array[32, byte].example])
     let emptyHandler = proc(log: Log) = discard
-
+    let id = await subscriptions.subscribeLogs(filter, emptyHandler)
-    check mockServer.newFilterCounter == 0
-    let jsonId = await subscriptions.subscribeLogs(filter, emptyHandler)
-    let id = string.fromJson(jsonId).tryGet
-    check mockServer.newFilterCounter == 1
-
-    await sleepAsync(50.millis)
     mockServer.invalidateFilter(id)
-    check eventually mockServer.newFilterCounter == 2
+    check eventually subscriptions.subscriptionMapping[id] != id
-    check mockServer.filters[id] == false
-    check mockServer.filters.len() == 2
-    await subscriptions.unsubscribe(jsonId)
-    check mockServer.filters.len() == 1
 
-    # invalidateFilter sets the filter's value to false which will return the "filter not found"
+    await subscriptions.unsubscribe(id)
-    # unsubscribing will actually delete the key from filters table
+
-    # hence after unsubscribing the only key left in the table should be the original id
+    check not subscriptions.filters.hasKey id
-    for key in mockServer.filters.keys():
+    check not subscriptions.subscriptionMapping.hasKey id
-      check key == id

testmodule/testContracts.nim

@@ -1,4 +1,5 @@
 import std/json
+import std/os
 import std/options
 import pkg/asynctest
 import pkg/questionable
@@ -16,7 +17,8 @@ type
 method mint(token: TestToken, holder: Address, amount: UInt256): Confirmable {.base, contract.}
 method myBalance(token: TestToken): UInt256 {.base, contract, view.}
 
-for url in ["ws://localhost:8545", "http://localhost:8545"]:
+let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
+for url in ["ws://"  & providerUrl, "http://"  & providerUrl]:
 
   suite "Contracts (" & url & ")":
 

testmodule/testCustomErrors.nim

@@ -1,3 +1,4 @@
+import std/os
 import std/json
 import pkg/asynctest
 import pkg/ethers
@@ -22,9 +23,10 @@ suite "Contract custom errors":
   var contract: TestCustomErrors
   var provider: JsonRpcProvider
   var snapshot: JsonNode
+  let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
 
   setup:
-    provider = JsonRpcProvider.new()
+    provider = JsonRpcProvider.new("http://" & providerUrl)
     snapshot = await provider.send("evm_snapshot")
     let deployment = readDeployment()
     let address = !deployment.address(TestCustomErrors)

testmodule/testEnums.nim

@@ -1,3 +1,4 @@
+import std/os
 import pkg/asynctest
 import pkg/ethers
 import pkg/serde
@@ -14,9 +15,10 @@ suite "Contract enum parameters and return values":
   var contract: TestEnums
   var provider: JsonRpcProvider
   var snapshot: JsonNode
+  let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
 
   setup:
-    provider = JsonRpcProvider.new()
+    provider = JsonRpcProvider.new("http://" & providerUrl)
     snapshot = await provider.send("evm_snapshot")
     let deployment = readDeployment()
     contract = TestEnums.new(!deployment.address(TestEnums), provider)

testmodule/testErc20.nim

@@ -1,3 +1,4 @@
+import std/os
 import std/json
 import pkg/asynctest
 import pkg/questionable
@@ -11,7 +12,8 @@ type
 
 method mint(token: TestToken, holder: Address, amount: UInt256): Confirmable {.base, contract.}
 
-for url in ["ws://localhost:8545", "http://localhost:8545"]:
+let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
+for url in ["ws://" & providerUrl, "http://"  & providerUrl]:
 
   suite "ERC20 (" & url & ")":
 

testmodule/testGasEstimation.nim

@@ -1,3 +1,4 @@
+import std/os
 import pkg/asynctest
 import pkg/ethers
 import pkg/serde
@@ -14,9 +15,10 @@ suite "gas estimation":
   var contract: TestGasEstimation
   var provider: JsonRpcProvider
   var snapshot: JsonNode
+  let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
 
   setup:
-    provider = JsonRpcProvider.new()
+    provider = JsonRpcProvider.new("http://" & providerUrl)
     snapshot = await provider.send("evm_snapshot")
     let deployment = readDeployment()
     let signer = provider.getSigner()

testmodule/testReturns.nim

@@ -1,3 +1,4 @@
+import std/os
 import pkg/asynctest
 import pkg/ethers
 import pkg/serde
@@ -13,9 +14,10 @@ suite "Contract return values":
   var contract: TestReturns
   var provider: JsonRpcProvider
   var snapshot: JsonNode
+  let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
 
   setup:
-    provider = JsonRpcProvider.new()
+    provider = JsonRpcProvider.new("http://" & providerUrl)
     snapshot = await provider.send("evm_snapshot")
     let deployment = readDeployment()
     contract = TestReturns.new(!deployment.address(TestReturns), provider)

testmodule/testTesting.nim

@@ -1,3 +1,4 @@
+import std/os
 import std/strformat
 import pkg/asynctest
 import pkg/chronos
@@ -92,9 +93,10 @@ suite "Testing helpers - contracts":
   var snapshot: JsonNode
   var accounts: seq[Address]
   let revertReason = "revert reason"
+  let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
 
   setup:
-    provider = JsonRpcProvider.new("ws://127.0.0.1:8545")
+    provider = JsonRpcProvider.new("ws://" & providerUrl)
     snapshot = await provider.send("evm_snapshot")
     accounts = await provider.listAccounts()
     helpersContract = TestHelpers.new(provider.getSigner())

testmodule/testWallet.nim

@@ -1,3 +1,4 @@
+import std/os
 import pkg/asynctest
 import pkg/serde
 import pkg/stew/byteutils
@@ -12,9 +13,10 @@ proc transfer*(erc20: Erc20, recipient: Address, amount: UInt256) {.contract.}
 suite "Wallet":
   var provider: JsonRpcProvider
   var snapshot: JsonNode
+  let providerUrl = getEnv("ETHERS_TEST_PROVIDER", "localhost:8545")
 
   setup:
-    provider = JsonRpcProvider.new()
+    provider = JsonRpcProvider.new("http://" & providerUrl)
     snapshot = await provider.send("evm_snapshot")
 
   teardown: