add toggle_private_query to libstorage

library/libstorage.h

@@ -218,6 +218,20 @@ extern "C"
         StorageCallback callback,
         void *userData);
 
+    // When set to true, runs all of the subsequent DHT **queries** over
+    // the Logos mix network. Note that this affects queries only, not
+    // advertisements.
+    //
+    // This is a **temporary** API and will likely be gone by mainnet.
+    //
+    // The callback returns a string containing the previous value for
+    // private queries ("true" if the were enabled, or "false" otherwise).
+    int storage_toggle_private_queries(
+        void *ctx,
+        bool enabled,
+        StorageCallback callback,
+        void *userData);
+
     // Initialize a download for `cid`.
     // `chunkSize` defines the size of each chunk to be used during download.
     // The default value is the default block size 1024 * 64 bytes.

library/libstorage.nim

@@ -37,6 +37,7 @@ import ./storage_thread_requests/requests/node_p2p_request
 import ./storage_thread_requests/requests/node_upload_request
 import ./storage_thread_requests/requests/node_download_request
 import ./storage_thread_requests/requests/node_storage_request
+import ./storage_thread_requests/requests/node_mix_request
 import ./ffi_types
 
 from ../storage/conf import storageVersion
@@ -361,6 +362,20 @@ proc storage_upload_file(
 
   return callback.okOrError(res, userData)
 
+proc storage_toggle_private_queries(
+    ctx: ptr StorageContext, enabled: bool, callback: StorageCallback, userData: pointer
+): cint {.dynlib, exportc.} =
+  initializeLibrary()
+  checkLibstorageParams(ctx, callback, userData)
+
+  let req = NodeMixRequest.createShared(privateQueries = enabled)
+
+  let res = storage_context.sendRequestToStorageThread(
+    ctx, RequestType.MIX, req, callback, userData
+  )
+
+  return callback.okOrError(res, userData)
+
 proc storage_download_init(
     ctx: ptr StorageContext,
     cid: cstring,

library/storage_thread_requests/requests/node_mix_request.nim

@@ -0,0 +1,25 @@
+import chronos
+import chronicles
+import results
+
+import ../../../storage/[storage, node]
+
+logScope:
+  topics = "libstorage libstoragemix"
+
+type NodeMixRequest* = object
+  privateQueries: bool
+
+proc createShared*(T: type NodeMixRequest, privateQueries: bool): ptr type T =
+  var ret = createShared(T)
+  ret[].privateQueries = privateQueries
+  return ret
+
+proc destroyShared(self: ptr NodeMixRequest) =
+  deallocShared(self)
+
+proc process*(
+    self: ptr NodeMixRequest, storage: ptr StorageServer
+): Future[Result[string, string]] {.async: (raises: []).} =
+  let previous = storage[].node.togglePrivateQueries(self.privateQueries)
+  return ok($previous)

library/storage_thread_requests/storage_thread_request.nim

@@ -13,6 +13,7 @@ import ./requests/node_p2p_request
 import ./requests/node_upload_request
 import ./requests/node_download_request
 import ./requests/node_storage_request
+import ./requests/node_mix_request
 
 from ../../storage/storage import StorageServer
 
@@ -24,6 +25,7 @@ type RequestType* {.pure.} = enum
   UPLOAD
   DOWNLOAD
   STORAGE
+  MIX
 
 type StorageThreadRequest* = object
   reqType: RequestType
@@ -122,6 +124,8 @@ proc process*(
       cast[ptr NodeUploadRequest](request[].reqContent).process(
         storage, onBlockReceived
       )
+    of MIX:
+      cast[ptr NodeMixRequest](request[].reqContent).process(storage)
 
   handleRes(await retFut, request)
 

storage/discovery.nim

@@ -50,6 +50,7 @@ type Discovery* = ref object of RootObj
   store: Datastore
   mixProto*: MixProtocol
   dhtMixProxies*: seq[SignedPeerRecord]
+  privateQueries: bool
 
 proc toNodeId*(cid: Cid): NodeId =
   ## Cid to discovery id
@@ -86,7 +87,7 @@ proc findPeer*(
 
   return PeerRecord.none
 
-proc findViaMix(
+method findViaMix(
     d: Discovery, cid: Cid
 ): Future[?!seq[SignedPeerRecord]] {.async: (raises: [CancelledError]).} =
   var candidates = d.dhtMixProxies
@@ -102,7 +103,7 @@ proc findViaMix(
 
   failure("All Mix lookup proxies failed (candidates=" & $candidates.len & ")")
 
-proc findDirect*(
+method findDirect*(
     d: Discovery, cid: Cid
 ): Future[?!seq[SignedPeerRecord]] {.async: (raises: [CancelledError]).} =
   try:
@@ -116,7 +117,7 @@ method find*(
     d: Discovery, cid: Cid
 ): Future[seq[SignedPeerRecord]] {.async: (raises: [CancelledError]), base.} =
   let providers =
-    if not d.mixProto.isNil and d.dhtMixProxies.len > 0:
+    if d.privateQueries and not d.mixProto.isNil and d.dhtMixProxies.len > 0:
       (await d.findViaMix(cid)).valueOr:
         warn "Mix lookup failed", cid, err = error.msg
         return @[]
@@ -261,6 +262,11 @@ proc close*(d: Discovery) {.async: (raises: []).} =
   else:
     trace "Discovery store closed"
 
+proc togglePrivateQueries*(d: Discovery, enabled: bool): bool =
+  let old = d.privateQueries
+  d.privateQueries = enabled
+  return old
+
 proc new*(
     T: type Discovery,
     key: PrivateKey,

storage/node.nim

@@ -446,6 +446,9 @@ proc iterateManifests*(self: StorageNodeRef, onManifest: OnManifest) {.async.} =
 
       onManifest(cid, manifest)
 
+proc togglePrivateQueries*(self: StorageNodeRef, enable: bool): bool =
+  return self.discovery.togglePrivateQueries(enable)
+
 proc onExpiryUpdate(
     self: StorageNodeRef, rootCid: Cid, expiry: SecondsSince1970
 ): Future[?!void] {.async: (raises: [CancelledError]).} =

tests/cbindings/storage.c

@@ -801,6 +801,43 @@ int check_delete(void *storage_ctx, const char *cid)
     return is_resp_ok(r, NULL);
 }
 
+int check_toggle_private_queries(void *storage_ctx)
+{
+    Resp *r = alloc_resp();
+    char *res = NULL;
+    // First toggle is false -> true
+    if (storage_toggle_private_queries(storage_ctx, true, (StorageCallback)callback, r) != RET_OK)
+    {
+        free_resp(r);
+        return RET_ERR;
+    }
+
+    int ret = is_resp_ok(r, &res);
+    printf("B\n");
+    if (strcmp(res, "false") != 0)
+    {
+        fprintf(stderr, "toggle private queries content mismatch, res:%s\n", res);
+        return RET_ERR;
+    }
+
+    // Second toggle is true -> false
+    r = alloc_resp();
+    if (storage_toggle_private_queries(storage_ctx, false, (StorageCallback)callback, r) != RET_OK)
+    {
+        free_resp(r);
+        return RET_ERR;
+    }
+
+    ret = is_resp_ok(r, &res);
+    if (strcmp(res, "true") != 0)
+    {
+        fprintf(stderr, "toggle private queries content mismatch, res:%s\n", res);
+        return RET_ERR;
+    }
+
+    return RET_OK;
+}
+
 // TODO: implement check_fetch
 // It is a bit complicated because it requires two nodes
 // connected together to fetch from peers.
@@ -852,6 +889,7 @@ int main(void)
 
     free(cid);
 
+    RUN_TEST(check_toggle_private_queries(storage_ctx));
     RUN_TEST(update_log_level(storage_ctx, "TRACE"));
     RUN_TEST(cleanup(storage_ctx));
 

tests/storage/blockexchange/discovery/testdiscovery.nim

@@ -1,7 +1,9 @@
 import std/[options, sequtils, sugar, tables]
 
 import pkg/chronos
+import pkg/libp2p/protocols/mix
 
+import pkg/storage/discovery
 import pkg/storage/rng
 import pkg/storage/stores
 import pkg/storage/blockexchange
@@ -129,3 +131,23 @@ asyncchecksuite "Block Advertising and Discovery":
     await engine.start()
     await sleepAsync(3.seconds)
     await engine.stop()
+
+  test "should route queries over mix when privacy toggle is enabled":
+    let
+      privateSpr = SignedPeerRecord.example
+      directSpr = SignedPeerRecord.example
+      refCid = Cid.example
+
+    let
+      # this is an invalid object, but we just need it to be non-null
+      mix = MixProtocol()
+      discovery = MixMockDiscovery.new()
+
+    discovery.mixProto = mix
+    discovery.privateSpr = privateSpr
+    discovery.directSpr = directSpr
+    discovery.refCid = refCid
+
+    check (await discovery.find(refCid)) == @[directSpr]
+    check discovery.togglePrivateQueries(true) == false
+    check (await discovery.find(refCid)) == @[privateSpr]

tests/storage/examples.nim

@@ -15,8 +15,11 @@ proc example*(_: type bt.Block, size: int = 4096): bt.Block =
   let bytes = newSeqWith(size, rand(uint8))
   bt.Block.new(bytes).tryGet()
 
+proc example*(_: type PrivateKey): PrivateKey =
+  PrivateKey.random(storage_rng.Rng.instance().libp2pRng).get
+
 proc example*(_: type PeerId): PeerId =
-  let key = PrivateKey.random(storage_rng.Rng.instance().libp2pRng).get
+  let key = PrivateKey.example
   PeerId.init(key.getPublicKey().get).get
 
 proc example*(_: type PeerContext): PeerContext =
@@ -25,6 +28,15 @@ proc example*(_: type PeerContext): PeerContext =
 proc example*(_: type Cid): Cid =
   bt.Block.example.cid
 
+proc example*(_: type SignedPeerRecord): SignedPeerRecord =
+  let
+    key = PrivateKey.example
+    peerId = PeerId.init(key.getPublicKey().get).get
+    record = PeerRecord.init(peerId, @[])
+    spr = SignedPeerRecord.init(key, record).tryGet()
+
+  spr
+
 proc example*(_: type BlockAddress): BlockAddress =
   BlockAddress.init(Cid.example, 0)
 

tests/storage/helpers/mockdiscovery.nim

@@ -10,6 +10,7 @@
 import pkg/chronos
 import pkg/libp2p
 import pkg/questionable
+import pkg/questionable/results
 import pkg/storage/discovery
 import pkg/contractabi/address as ca
 
@@ -98,3 +99,23 @@ proc nullDiscovery*(): MockDiscovery =
     findHostProvidersHandler: findHostProvidersHandler,
     publishHostProvideHandler: publishHostProvideHandler,
   )
+
+# Slightly more contrived Discovery mock to allow testing of the privacy toggle.
+# Since we cannot declare `method` within blocks, we have to do this contortionism
+# here.
+type MixMockDiscovery* = ref object of Discovery
+  privateSpr*: SignedPeerRecord
+  directSpr*: SignedPeerRecord
+  refCid*: Cid
+
+method findViaMix*(
+    d: MixMockDiscovery, cid: Cid
+): Future[?!seq[SignedPeerRecord]] {.async: (raises: [CancelledError]).} =
+  doAssert cid == d.refCid
+  result = success(@[d.privateSpr])
+
+method findDirect*(
+    d: MixMockDiscovery, cid: Cid
+): Future[?!seq[SignedPeerRecord]] {.async: (raises: [CancelledError]).} =
+  doAssert cid == d.refCid
+  result = success(@[d.directSpr])