GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Max/mixfetch concurrent test (#6417)

Browse Source 
* * Experiment with changing address mapping from canonical -> full URL as
  string.
* Up MaxConns config.

* Bump webpack-cli version

* Modify internal-dev tester for concurrent testing

* Add logging + POST request to internal-dev/ 

* push lockfiles

* Remove RequestURL from RequestOptions struct for interface

* Bump versions + update lockfiles
This commit is contained in:
mfahampshire
2026-02-23 15:30:49 +00:00
committed by GitHub
parent 6edbece3ad
commit 630c4922ac
15 changed files with 1910 additions and 1016 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -1
View File
@@ -4954,7 +4954,7 @@ dependencies = [
[[package]]
name = "mix-fetch-wasm"
version = "1.4.1"
version = "1.4.2"
dependencies = [
"async-trait",
"futures",
sdk/typescript/packages/mix-fetch-node/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@nymproject/mix-fetch-node",
"version": "1.4.1",
"version": "1.4.2",
"description": "This package is a drop-in replacement for `fetch` in NodeJS to send HTTP requests over the Nym Mixnet.",
"license": "Apache-2.0",
"author": "Nym Technologies SA",
sdk/typescript/packages/mix-fetch/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@nymproject/mix-fetch",
"version": "1.4.1",
"version": "1.4.2",
"description": "This package is a drop-in replacement for `fetch` to send HTTP requests over the Nym Mixnet.",
"license": "Apache-2.0",
"author": "Nym Technologies SA",
wasm/mix-fetch/Cargo.toml
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
name = "mix-fetch-wasm"
authors = ["Jedrzej Stuczynski <andrew@nymtech.net>"]
version = "1.4.1"
version = "1.4.2"
edition = "2021"
keywords = ["nym", "fetch", "wasm", "webassembly", "privacy"]
license = "Apache-2.0"
wasm/mix-fetch/Makefile
+1
View File
@@ -10,6 +10,7 @@ build-go-opt:
build-rust:
taskset -c 0-11 wasm-pack build --scope nymproject --target web --out-dir ../../dist/wasm/mix-fetch
# taskset -c 0-11 wasm-pack build --scope nymproject --target no-modules --out-dir ../../dist/wasm/mix-fetch
taskset -c 0-11 wasm-opt -Oz -o ../../dist/wasm/mix-fetch/mix_fetch_wasm_bg.wasm ../../dist/wasm/mix-fetch/mix_fetch_wasm_bg.wasm
build-rust-debug:
wasm/mix-fetch/go-mix-conn/internal/mixfetch/mixfetch.go
+34 -21
View File
@@ -142,7 +142,7 @@ func schemeFetch(req *conv.ParsedRequest) error {
}
}
func dialContext(_ctx context.Context, opts *types.RequestOptions, _network, addr string) (net.Conn, error) {
func dialContext(_ctx context.Context, requestURL string, _network, addr string) (net.Conn, error) {
log.Debug("dialing plain connection to %s", addr)
requestId, err := rust_bridge.RsStartNewMixnetRequest(addr)
@@ -154,12 +154,14 @@ func dialContext(_ctx context.Context, opts *types.RequestOptions, _network, add
}
conn, inj := state.NewFakeConnection(requestId, addr)
state.ActiveRequests.Insert(requestId, addr, inj)
// Use requestURL (full URL) as the mapping key, meaning we can now
// have concurrent requests to different paths on the same domain.
state.ActiveRequests.Insert(requestId, requestURL, inj)
return conn, nil
}
func dialTLSContext(_ctx context.Context, opts *types.RequestOptions, _network, addr string) (net.Conn, error) {
func dialTLSContext(_ctx context.Context, requestURL string, _network, addr string) (net.Conn, error) {
log.Debug("dialing TLS connection to %s", addr)
requestId, err := rust_bridge.RsStartNewMixnetRequest(addr)
@@ -171,7 +173,9 @@ func dialTLSContext(_ctx context.Context, opts *types.RequestOptions, _network,
}
conn, inj := state.NewFakeTlsConn(requestId, addr)
state.ActiveRequests.Insert(requestId, addr, inj)
// Use requestURL (full URL) as the mapping key, meaning we can now
// have concurrent requests to different paths on the same domain.
state.ActiveRequests.Insert(requestId, requestURL, inj)
if err := conn.Handshake(); err != nil {
return nil, err
@@ -180,7 +184,7 @@ func dialTLSContext(_ctx context.Context, opts *types.RequestOptions, _network,
return conn, nil
}
func buildHttpClient(reqCtx *types.RequestContext, opts *types.RequestOptions) *http.Client {
func buildHttpClient(reqCtx *types.RequestContext, opts *types.RequestOptions, requestURL string) *http.Client {
return &http.Client{
CheckRedirect: func(req *http.Request, via []*http.Request) error {
return checkRedirect(reqCtx, opts, req, via)
@@ -188,17 +192,19 @@ func buildHttpClient(reqCtx *types.RequestContext, opts *types.RequestOptions) *
Transport: &http.Transport{
DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
return dialContext(ctx, opts, network, addr)
return dialContext(ctx, requestURL, network, addr)
},
DialTLSContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
return dialTLSContext(ctx, opts, network, addr)
return dialTLSContext(ctx, requestURL, network, addr)
},
//TLSClientConfig: &tlsConfig,
DisableKeepAlives: true,
MaxIdleConns: 1,
MaxIdleConnsPerHost: 1,
MaxConnsPerHost: 1,
DisableKeepAlives: true,
// Allow multiple concurrent connections to the same host.
// Previously set to 1.
MaxIdleConns: 10,
MaxIdleConnsPerHost: 10,
MaxConnsPerHost: 10,
},
Timeout: state.RequestTimeout,
}
@@ -270,7 +276,7 @@ func doCorsCheck(reqOpts *types.RequestOptions, resp *http.Response) error {
return errors.New("failed cors check")
}
func performRequest(req *conv.ParsedRequest) (*conv.ResponseWrapper, error) {
func performRequest(req *conv.ParsedRequest, requestURL string) (*conv.ResponseWrapper, error) {
err := mainFetchChecks(req)
if err != nil {
return nil, err
@@ -278,7 +284,7 @@ func performRequest(req *conv.ParsedRequest) (*conv.ResponseWrapper, error) {
reqCtx := &types.RequestContext{}
reqClient := buildHttpClient(reqCtx, req.Options)
reqClient := buildHttpClient(reqCtx, req.Options, requestURL)
if req.Options.ReferrerPolicy == "" {
// 4.1.8
@@ -322,12 +328,15 @@ func performRequest(req *conv.ParsedRequest) (*conv.ResponseWrapper, error) {
func onErrCleanup(url *url.URL) {
// TODO: cancel stuff here.... somehow...
canonicalAddr := canonicalAddr(url)
id := state.ActiveRequests.GetId(canonicalAddr)
// Use full URL string to match the key used in MixFetch for request deduplication.
// Makes sure we clean up the correct request when multiple requests to
// different paths on the same domain are in process.
requestURL := url.String()
id := state.ActiveRequests.GetId(requestURL)
// TODO: can we guarantee that rust is not holding any references to that id (that we don't know on this side)?
if id == 0 {
// if id doesn't exist it [probably] means the error was thrown before the request was properly created
log.Debug("there doesn't seem to exist a request associated with addr %s", canonicalAddr)
log.Debug("there doesn't seem to exist a request associated with URL %s", requestURL)
return
}
state.ActiveRequests.Remove(id)
@@ -341,16 +350,20 @@ func onErrCleanup(url *url.URL) {
func MixFetch(request *conv.ParsedRequest) (any, error) {
log.Info("_mixFetch: start")
canonical := canonicalAddr(request.Request.URL)
if state.ActiveRequests.ExistsCanonical(canonical) {
// TODO: how to deal with it to allow for concurrent requests to say `https://foo.com/index.html` and `https://foo.com/index.js`?
return nil, errors.New(fmt.Sprintf("there is already an active request for %s", canonical))
// Use the full URL (inc path and query params) as the deduplication key.
// Allows concurrent requests to different paths on the same domain
// (e.g., foo.com/index.html and foo.com/index.js) while still preventing
// duplicate requests to the exact same URL.
requestURL := request.Request.URL.String()
if state.ActiveRequests.ExistsCanonical(requestURL) {
return nil, errors.New(fmt.Sprintf("there is already an active request for %s", requestURL))
}
resCh := make(chan *conv.ResponseWrapper)
errCh := make(chan error)
go func(resCh chan *conv.ResponseWrapper, errCh chan error) {
resp, err := performRequest(request)
resp, err := performRequest(request, requestURL)
if err != nil {
errCh <- err
} else {
wasm/mix-fetch/go-mix-conn/internal/state/state.go
+28 -17
View File
@@ -25,17 +25,23 @@ func InitialiseGlobalState() {
}
}
// CurrentActiveRequests tracks ongoign requests for thread-safe access.
// The AddressMapping uses full URLs (inc path and query params) as keys,
// allowing concurrent requests to different paths on the same domain while
// preventing duplicate requests to the *exact* same URL.
type CurrentActiveRequests struct {
sync.Mutex
Requests map[types.RequestId]*ActiveRequest
AddressMapping map[string]types.RequestId
AddressMapping map[string]types.RequestId // key is full URL string
}
func (ar *CurrentActiveRequests) GetId(canonicalAddr string) types.RequestId {
log.Trace("getting id associated with request for %s", canonicalAddr)
// GetId returns the request ID associated with the given URL string.
// The URL should be the full URL including path and query params.
func (ar *CurrentActiveRequests) GetId(requestURL string) types.RequestId {
log.Trace("getting id associated with request for %s", requestURL)
ar.Lock()
defer ar.Unlock()
return ar.AddressMapping[canonicalAddr]
return ar.AddressMapping[requestURL]
}
func (ar *CurrentActiveRequests) Exists(id types.RequestId) bool {
@@ -46,25 +52,30 @@ func (ar *CurrentActiveRequests) Exists(id types.RequestId) bool {
return exists
}
func (ar *CurrentActiveRequests) ExistsCanonical(canonicalAddr string) bool {
return ar.GetId(canonicalAddr) != 0
// ExistsCanonical checks if there's already an active request for the given URL.
// The URL should be the full URL including path and query params.
// Allows concurrent requests to different paths on the same domain.
func (ar *CurrentActiveRequests) ExistsCanonical(requestURL string) bool {
return ar.GetId(requestURL) != 0
}
func (ar *CurrentActiveRequests) Insert(id types.RequestId, canonicalAddr string, inj ConnectionInjector) {
log.Trace("inserting request %d for %s", id, canonicalAddr)
// Insert adds a new active request to the tracking maps.
// The requestURL should be the full URL including path and query params.
func (ar *CurrentActiveRequests) Insert(id types.RequestId, requestURL string, inj ConnectionInjector) {
log.Trace("inserting request %d for %s", id, requestURL)
ar.Lock()
defer ar.Unlock()
_, exists := ar.Requests[id]
if exists {
panic("attempted to overwrite active connection id")
}
_, exists = ar.AddressMapping[canonicalAddr]
_, exists = ar.AddressMapping[requestURL]
if exists {
panic("attempted to overwrite active connection canonicalAddr")
panic("attempted to overwrite active connection for URL")
}
ar.Requests[id] = &ActiveRequest{injector: inj, canonicalAddr: canonicalAddr}
ar.AddressMapping[canonicalAddr] = id
ar.Requests[id] = &ActiveRequest{injector: inj, requestURL: requestURL}
ar.AddressMapping[requestURL] = id
}
func (ar *CurrentActiveRequests) Remove(id types.RequestId) {
@@ -75,13 +86,13 @@ func (ar *CurrentActiveRequests) Remove(id types.RequestId) {
if !exists {
panic("attempted to remove active connection id that doesn't exist")
}
_, exists = ar.AddressMapping[req.canonicalAddr]
_, exists = ar.AddressMapping[req.requestURL]
if !exists {
panic("attempted to remove active connection canonicalAddr that doesn't exist")
panic("attempted to remove active connection URL that doesn't exist")
}
delete(ar.Requests, id)
delete(ar.AddressMapping, req.canonicalAddr)
delete(ar.AddressMapping, req.requestURL)
}
func (ar *CurrentActiveRequests) InjectData(id types.RequestId, data []byte) {
@@ -119,6 +130,6 @@ func (ar *CurrentActiveRequests) SendError(id types.RequestId, err error) {
}
type ActiveRequest struct {
injector ConnectionInjector
canonicalAddr string
injector ConnectionInjector
requestURL string // Full URL including path and query params
}
wasm/mix-fetch/internal-dev/index.html
+93 -18
View File
@@ -1,27 +1,102 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Nym MixFetch Demo</title>
<script src="bootstrap.js"></script>
</head>
</head>
<body>
<h1> Mix Fetch Demo</h1>
<div>
<label>Target Host: </label>
<input type="text" size = "60" id="fetch_payload" value="...">
<button id="fetch-button"> Fetch </button>
</div>
<body>
<h1>Mix Fetch Demo</h1>
<fieldset id="startup-controls">
<legend>MixFetch Configuration</legend>
<p>
You can either use the default Gateway/NR combo run by us, or have
MixFetch choose a random one on startup.
</p>
<div style="margin-bottom: 10px">
<label>
<input type="radio" name="gateway-mode" value="default" checked />
Default Gateway (q2A2cbooyC16YJzvdYaSMH9X3cSiieZNtfBr8cE8Fi1)
</label>
</div>
<div style="margin-bottom: 10px">
<label>
<input type="radio" name="gateway-mode" value="random" />
Random Gateway
</label>
</div>
<button id="start-mixfetch">Start MixFetch</button>
<span id="mixfetch-status" style="margin-left: 10px; color: gray"
>Not started</span
>
</fieldset>
<hr>
<p>
<span id="output"></span>
</p>
<hr />
</body>
<fieldset id="fetch-controls" disabled>
<legend>Fetch Controls</legend>
<div>
<label>Target Host 1: </label>
<input
type="text"
size="60"
id="fetch_payload_1"
value="https://api.ipify.org?format=json"
/>
<button id="fetch-button-1">Fetch 1</button>
</div>
<div>
<label>Target Host 2: </label>
<input
type="text"
size="60"
id="fetch_payload_2"
value="https://api6.ipify.org?format=json"
/>
<button id="fetch-button-2">Fetch 2</button>
</div>
<p>
Note: if you're hammering these endpoints and you start to get timeouts
(or you've been reloading the page a lot) then change the target hosts
to e.g. http://ipv4.icanhazip.com and https://ipv6.icanhazip.com/
</p>
<div>
<button id="fetch-10-concurrent">
Send 10 Concurrent Requests (posts/1-10)
</button>
<p>
This does what is says on the tin - sends 10 fetch requests to
https://jsonplaceholder.typicode.com/posts/ 1 through 10
</p>
</div>
<hr />
<div>
<label>POST URL: </label>
<input
type="text"
size="60"
id="post_url"
value="https://jsonplaceholder.typicode.com/posts"
/>
</div>
<div>
<label>POST Body (JSON): </label>
<textarea id="post_body" rows="3" cols="60">
{"title": "Test Post", "body": "Hello from MixFetch!", "userId": 1}</textarea
>
</div>
<div>
<button id="post-button">Send POST Request</button>
</div>
<p>Do a POST and get it echoed back</p>
</fieldset>
</html>
<hr />
<p>
<span id="output"></span>
</p>
</body>
</html>
wasm/mix-fetch/internal-dev/index.js
+146 -6
View File
@@ -25,11 +25,36 @@ class WebWorkerClient {
const { rawString } = ev.data.args;
displayReceivedRawString(rawString)
break;
case 'Log':
const { message, level } = ev.data.args;
displayLog(message, level);
break;
case 'MixFetchReady':
onMixFetchReady();
break;
case 'MixFetchError':
const { error } = ev.data.args;
onMixFetchError(error);
break;
}
}
};
}
startMixFetch = (preferredGateway) => {
if (!this.worker) {
console.error('Could not send message because worker does not exist');
return;
}
this.worker.postMessage({
kind: 'StartMixFetch',
args: {
preferredGateway,
},
});
}
doFetch = (target) => {
if (!this.worker) {
console.error('Could not send message because worker does not exist');
@@ -43,25 +68,140 @@ class WebWorkerClient {
},
});
}
doPost = (url, body) => {
if (!this.worker) {
console.error('Could not send message because worker does not exist');
return;
}
this.worker.postMessage({
kind: 'PostPayload',
args: {
url,
body,
},
});
}
}
let client = null;
const DEFAULT_GATEWAY = "q2A2cbooyC16YJzvdYaSMH9X3cSiieZNtfBr8cE8Fi1";
async function main() {
client = new WebWorkerClient();
const fetchButton = document.querySelector('#fetch-button');
fetchButton.onclick = function () {
doFetch();
const startButton = document.querySelector('#start-mixfetch');
startButton.onclick = function () {
const gatewayMode = document.querySelector('input[name="gateway-mode"]:checked').value;
const preferredGateway = gatewayMode === 'default' ? DEFAULT_GATEWAY : undefined;
startButton.disabled = true;
document.querySelectorAll('input[name="gateway-mode"]').forEach(r => r.disabled = true);
updateStatus('Starting...', 'orange');
displayLog(`Starting MixFetch with ${gatewayMode} gateway${preferredGateway ? ` (${preferredGateway})` : ''}...`, 'info');
client.startMixFetch(preferredGateway);
}
const fetchButton1 = document.querySelector('#fetch-button-1');
fetchButton1.onclick = function () {
doFetch(1);
}
const fetchButton2 = document.querySelector('#fetch-button-2');
fetchButton2.onclick = function () {
doFetch(2);
}
const fetch10Button = document.querySelector('#fetch-10-concurrent');
fetch10Button.onclick = function () {
doFetch10Concurrent();
}
const postButton = document.querySelector('#post-button');
postButton.onclick = function () {
doPost();
}
}
function updateStatus(text, color) {
const status = document.getElementById('mixfetch-status');
status.textContent = text;
status.style.color = color;
}
async function doFetch() {
const payload = document.getElementById('fetch_payload').value;
function onMixFetchReady() {
updateStatus('Ready', 'green');
document.getElementById('fetch-controls').disabled = false;
displayLog('MixFetch is ready!', 'info');
}
function onMixFetchError(error) {
updateStatus('Error: ' + error, 'red');
document.querySelector('#start-mixfetch').disabled = false;
document.querySelectorAll('input[name="gateway-mode"]').forEach(r => r.disabled = false);
displayLog('MixFetch error: ' + error, 'error');
}
async function doFetch(id) {
const payload = document.getElementById(`fetch_payload_${id}`).value;
await client.doFetch(payload)
displaySend(`clicked the button and the payload is: ${payload}...`);
displaySend(`[${id}] clicked the button and the payload is: ${payload}...`);
}
async function doFetch10Concurrent() {
const baseUrl = 'https://jsonplaceholder.typicode.com/posts/';
displaySend('Starting 10 concurrent requests to posts/1-10...');
const requests = [];
for (let i = 1; i <= 10; i++) {
const url = `${baseUrl}${i}`;
displaySend(`[${i}] Sending request to ${url}`);
requests.push(client.doFetch(url));
}
await Promise.all(requests);
displaySend('All 10 concurrent requests dispatched!');
}
async function doPost() {
const url = document.getElementById('post_url').value;
const body = document.getElementById('post_body').value;
displaySend(`[POST] Sending POST request to ${url}`);
displaySend(`[POST] Body: ${body}`);
await client.doPost(url, body);
}
/**
* Display log messages from MixFetch. Colors based on level.
*
* @param {string} message
* @param {string} level - 'info', 'error', 'warn', or 'debug'
*/
function displayLog(message, level) {
let timestamp = new Date().toISOString().substr(11, 12);
const colors = {
info: 'gray',
error: 'red',
warn: 'orange',
debug: 'purple',
};
let logDiv = document.createElement('div');
let paragraph = document.createElement('p');
paragraph.setAttribute('style', `color: ${colors[level] || 'gray'}`);
let paragraphContent = document.createTextNode(timestamp + ' [' + level.toUpperCase() + '] ' + message);
paragraph.appendChild(paragraphContent);
logDiv.appendChild(paragraph);
document.getElementById('output').appendChild(logDiv);
}
/**
wasm/mix-fetch/internal-dev/package-lock.json
Generated
+1111 -442
View File
File diff suppressed because it is too large Load Diff
wasm/mix-fetch/internal-dev/package.json
+2 -2
View File
@@ -30,8 +30,8 @@
"devDependencies": {
"copy-webpack-plugin": "^11.0.0",
"hello-wasm-pack": "^0.1.0",
"webpack": "^5.105.0",
"webpack-cli": "^4.9.2",
"webpack": "^5.98.0",
"webpack-cli": "^5.1.4",
"webpack-dev-server": "^5.2.1"
},
"dependencies": {
wasm/mix-fetch/internal-dev/webpack.config.js
+33 -34
View File
@@ -1,37 +1,36 @@
const CopyWebpackPlugin = require('copy-webpack-plugin');
const path = require('path');
const CopyWebpackPlugin = require("copy-webpack-plugin");
const path = require("path");
module.exports = {
performance: {
hints: false,
maxEntrypointSize: 512000,
maxAssetSize: 512000
},
entry: {
bootstrap: './bootstrap.js',
worker: './worker.js',
},
output: {
path: path.resolve(__dirname, 'dist'),
filename: '[name].js',
},
mode: 'development',
// mode: 'production',
plugins: [
new CopyWebpackPlugin({
patterns: [
'index.html',
{
from: '../pkg/*.(js|wasm)',
to: '[name][ext]',
},
{
from: '../go-mix-conn/build/*.(js|wasm)',
to: '[name][ext]',
},
],
}),
],
experiments: { syncWebAssembly: true },
performance: {
hints: false,
maxEntrypointSize: 512000,
maxAssetSize: 512000,
},
entry: {
bootstrap: "./bootstrap.js",
worker: "./worker.js",
},
output: {
path: path.resolve(__dirname, "dist"),
filename: "[name].js",
},
mode: "development",
// mode: 'production',
plugins: [
new CopyWebpackPlugin({
patterns: [
"index.html",
{
from: "../pkg/*.(js|wasm)",
to: "[name][ext]",
},
{
from: "../go-mix-conn/build/*.(js|wasm)",
to: "[name][ext]",
},
],
}),
],
experiments: { syncWebAssembly: true },
};
wasm/mix-fetch/internal-dev/worker.js
+117 -102
View File
@@ -43,6 +43,25 @@ let client = null;
let tester = null;
const go = new Go(); // Defined in wasm_exec.js
var goWasm;
let mixFetchReady = false;
function sendLog(message, level = 'info') {
self.postMessage({
kind: 'Log',
args: { message, level },
});
}
function sendReady() {
self.postMessage({ kind: 'MixFetchReady' });
}
function sendError(error) {
self.postMessage({
kind: 'MixFetchError',
args: { error: String(error) },
});
}
async function logFetchResult(res) {
console.log(res)
@@ -121,114 +140,117 @@ async function wasm_bindgenSetup() {
// await setupMixFetchWithConfig(config, { storagePassphrase: "foomp", preferredGateway })
}
async function nativeSetup() {
const preferredGateway = "8ookuLkA9oWfRTjb7Jq4tLGcWrqoXKGQxw84MjMrv2S4";
const validator = 'https://sandbox-nym-api1.nymtech.net/api';
async function nativeSetup(preferredGateway) {
sendLog('Setting up MixFetch...');
if (preferredGateway) {
sendLog(`Using preferred gateway: ${preferredGateway}`);
} else {
sendLog('Using random gateway selection');
}
// local
// const preferredNetworkRequester= "2o47bhnXWna6VEyt4mXMGQQAbXfpKmX7BkjkxUz8uQVi.6uQGnCqSczpXwh86NdbsCoDDXuqZQM9Uwko8GE7uC9g8@6qQYb4ArXANU6HJDxzH4PFCUqYb39Dae2Gem2KpxescM";
// const preferredNetworkRequester= "GqiGWmKRCbGQFSqH88BzLKijvZgipnqhmbNFsmkZw84t.4L8sXFuAUyUYyHZYgMdM3AtiusKnYUft6Pd8e41rrCHA@6qQYb4ArXANU6HJDxzH4PFCUqYb39Dae2Gem2KpxescM";
// those are just some examples, there are obviously more permutations;
// note, the extra optional argument is of the following type:
/*
export interface MixFetchOpts extends MixFetchOptsSimple {
clientId?: string;
nymApiUrl?: string;
nyxdUrl?: string;
clientOverride?: DebugWasmOverride;
mixFetchOverride?: MixFetchDebugOverride;
}
where
export interface MixFetchDebugOverride {
requestTimeoutMs?: number;
}
and `DebugWasmOverride` is a rather nested struct that you can look up yourself : )
*/
// #1
// await setupMixFetch(preferredNetworkRequester)
// #2
// await setupMixFetch(preferredNetworkRequester, { nymApiUrl: validator })
// // #3
const noCoverTrafficOverride = {
traffic: {disableMainPoissonPacketDistribution: true},
coverTraffic: {disableLoopCoverTrafficStream: true},
}
const mixFetchOverride = {
requestTimeoutMs: 20000
requestTimeoutMs: 60000
}
await setupMixFetch({
// preferredNetworkRequester,
// preferredGateway: preferredGateway,
// storagePassphrase: "foomp",
const opts = {
forceTls: true,
// nymApiUrl: validator,
clientId: "my-client",
clientOverride: noCoverTrafficOverride,
mixFetchOverride,
})
};
if (preferredGateway) {
opts.preferredGateway = preferredGateway;
}
sendLog('Calling setupMixFetch...');
await setupMixFetch(opts);
sendLog('setupMixFetch completed');
}
async function testMixFetch() {
console.log('Instantiating Mix Fetch...');
// await wasm_bindgenSetup()
async function startMixFetch(preferredGateway) {
sendLog('Instantiating MixFetch...');
await nativeSetup()
try {
await nativeSetup(preferredGateway);
mixFetchReady = true;
sendLog('MixFetch client running!');
sendReady();
} catch (e) {
sendLog('Failed to start MixFetch: ' + e, 'error');
sendError(e);
}
}
async function handleFetchPayload(target) {
if (!mixFetchReady) {
sendLog('MixFetch not ready yet', 'error');
return;
}
console.log('Mix Fetch client running!');
const url = target;
const args = {mode: "unsafe-ignore-cors"};
// Set callback to handle messages passed to the worker.
try {
sendLog(`Fetching: ${url}`);
const mixFetchRes = await mixFetch(url, args);
sendLog('Fetch completed');
await logFetchResult(mixFetchRes);
} catch (e) {
sendLog('Fetch request failure: ' + e, 'error');
console.error("mix fetch request failure: ", e);
}
}
async function handlePostPayload(url, body) {
if (!mixFetchReady) {
sendLog('MixFetch not ready yet', 'error');
return;
}
const args = {
method: 'POST',
mode: "unsafe-ignore-cors",
headers: {
'Content-Type': 'application/json',
},
body: body,
};
try {
sendLog(`POST request to: ${url}`);
sendLog(`POST body: ${body}`);
const mixFetchRes = await mixFetch(url, args);
sendLog('POST completed');
await logFetchResult(mixFetchRes);
} catch (e) {
sendLog('POST request failure: ' + e, 'error');
console.error("mix fetch POST request failure: ", e);
}
}
function setupMessageHandler() {
self.onmessage = async event => {
if (event.data && event.data.kind) {
switch (event.data.kind) {
case 'StartMixFetch': {
const { preferredGateway } = event.data.args;
await startMixFetch(preferredGateway);
break;
}
case 'FetchPayload': {
const {target} = event.data.args;
const url = target;
// const args = { mode: "ors", redirect: "manual", signal }
const args = {mode: "unsafe-ignore-cors"}
try {
console.log('using mixFetch...');
const mixFetchRes = await mixFetch(url, args)
console.log(">>> MIX FETCH")
await logFetchResult(mixFetchRes)
console.log('done')
} catch (e) {
console.error("mix fetch request failure: ", e)
}
// console.log("will disconnect");
// await disconnectMixFetch()
//
// try {
// console.log('using mixFetch...');
// const mixFetchRes = await mixFetch(url, args)
// console.log(">>> MIX FETCH")
// await logFetchResult(mixFetchRes)
//
// console.log('done')
//
// } catch(e) {
// console.error("mix fetch request failure: ", e)
// }
// try {
// console.log('using normal Fetch...');
// const fetchRes = await fetch(url, args)
// console.log(">>> NORMAL FETCH")
// await logFetchResult(fetchRes)
// } catch(e) {
// console.error("fetch request failure: ", e)
// }
const { target } = event.data.args;
await handleFetchPayload(target);
break;
}
case 'PostPayload': {
const { url, body } = event.data.args;
await handlePostPayload(url, body);
break;
}
}
}
@@ -263,15 +285,17 @@ function setupRsGoBridge() {
}
async function main() {
console.log(">>>>>>>>>>>>>>>>>>>>> JS WORKER MAIN START");
sendLog('Worker starting...');
// load rust WASM package
sendLog('Loading Rust WASM...');
await wasm_bindgen(RUST_WASM_URL);
console.log('Loaded RUST WASM');
sendLog('Loaded Rust WASM');
// load go WASM package
sendLog('Loading Go WASM...');
await loadGoWasm();
console.log("Loaded GO WASM");
sendLog('Loaded Go WASM');
// sets up better stack traces in case of in-rust panics
set_panic_hook();
@@ -280,19 +304,10 @@ async function main() {
goWasmSetLogging("trace")
// test mixFetch
await testMixFetch();
//
// // run test on simplified and dedicated tester:
// // await testWithTester()
//
// // hook-up the whole client for testing
// // await testWithNymClient()
//
// // 'Normal' client setup (to send 'normal' messages)
// // await normalNymClientUsage()
//
console.log(">>>>>>>>>>>>>>>>>>>>> JS WORKER MAIN END")
// Set up message handler (MixFetch will be started on demand)
setupMessageHandler();
sendLog('Worker ready - click Start MixFetch to begin');
}
// Let's get started!
wasm/mix-fetch/internal-dev/yarn.lock
+339 -370
View File
File diff suppressed because it is too large Load Diff
wasm/mix-fetch/src/client.rs
+2
View File
@@ -249,6 +249,8 @@ impl MixFetchClient {
request_id: RequestId,
target: RemoteAddress,
) -> Result<(), MixFetchError> {
console_log!(">>>>>>>>>>> SENDING CONNECT");
let raw_conn_req = socks5_connect_request(request_id, target, self.self_address);
let lane = TransmissionLane::ConnectionId(request_id);
let input = InputMessage::new_regular(