adding IsReady to dialer interface

chained/algeneva_impl.go

@@ -73,6 +73,10 @@ func (a *algenevaImpl) dialServer(op *ops.Op, ctx context.Context) (net.Conn, er
 	return conn, nil
 }
 
+func (*algenevaImpl) ready() <-chan error {
+	return nil
+}
+
 // algenevaDialer is a algeneva.Dialer wrapper around a reportDialCore. algeneva accepts an optional
 // Dialer interface which it will use to dial the server and then wrap the resulting connection.
 type algenevaDialer struct {

chained/broflake_impl.go

@@ -55,6 +55,10 @@ func newBroflakeImpl(pc *config.ProxyConfig, reportDialCore reportDialCoreFn) (p
 	}, nil
 }
 
+func (*broflakeImpl) ready() <-chan error {
+	return nil
+}
+
 func (b *broflakeImpl) dialServer(op *ops.Op, ctx context.Context) (net.Conn, error) {
 	// TODO: I don't know what to do with 'op'
 	return b.QUICLayer.DialContext(ctx)

chained/chained_test.go

@@ -51,6 +51,10 @@ func (impl *testImpl) dialServer(op *ops.Op, ctx context.Context) (net.Conn, err
 	return impl.d(ctx)
 }
 
+func (*testImpl) ready() <-chan error {
+	return nil
+}
+
 func newDialer(dialServer func(ctx context.Context) (net.Conn, error)) (func(network, addr string) (net.Conn, error), error) {
 	p, err := newProxy("test", "addr:567", "proto", "netw", &config.ProxyConfig{
 		AuthToken: "token",

chained/http_impl.go

@@ -22,3 +22,7 @@ func newHTTPImpl(addr string, dialCore coreDialer) proxyImpl {
 func (impl *httpImpl) dialServer(op *ops.Op, ctx context.Context) (net.Conn, error) {
 	return impl.dialCore(op, ctx, impl.addr)
 }
+
+func (*httpImpl) ready() <-chan error {
+	return nil
+}

chained/https_impl.go

@@ -106,6 +106,10 @@ func (impl *httpsImpl) dialServer(op *ops.Op, ctx context.Context) (net.Conn, er
 	return result.Conn, nil
 }
 
+func (*httpsImpl) ready() <-chan error {
+	return nil
+}
+
 func timeoutFor(ctx context.Context) time.Duration {
 	deadline, ok := ctx.Deadline()
 	if ok {

chained/multipath.go

@@ -49,6 +49,10 @@ func (impl *multipathImpl) FormatStats() []string {
 	return impl.dialer.(multipath.Stats).FormatStats()
 }
 
+func (*multipathImpl) ready() <-chan error {
+	return nil
+}
+
 func CreateMPDialer(configDir, endpoint string, ss map[string]*config.ProxyConfig, uc common.UserConfig) (dialer.ProxyDialer, error) {
 	if len(ss) < 1 {
 		return nil, errors.New("no dialers")

chained/multiplexed_impl.go

@@ -51,6 +51,10 @@ func (impl *multiplexedImpl) dialServer(op *ops.Op, ctx context.Context) (net.Co
 	return impl.multiplexedDial(ctx, "", "")
 }
 
+func (impl *multiplexedImpl) ready() <-chan error {
+	return nil
+}
+
 // createMultiplexedProtocol configures a cmux multiplexing protocol
 // according to settings.
 func createMultiplexedProtocol(s *config.ProxyConfig) (cmux.Protocol, error) {

chained/preconnecting_dialer.go

@@ -73,6 +73,10 @@ func (pd *preconnectingDialer) dialServer(op *ops.Op, ctx context.Context) (conn
 	}
 }
 
+func (*preconnectingDialer) ready() <-chan error {
+	return nil
+}
+
 func (pd *preconnectingDialer) preconnectIfNecessary(op *ops.Op) {
 	pd.statsMutex.Lock()
 	defer pd.statsMutex.Unlock()

chained/proxy.go

@@ -67,6 +67,8 @@ type proxyImpl interface {
 	dialServer(op *ops.Op, ctx context.Context) (net.Conn, error)
 	// close releases the resources associated with the implementation, if any.
 	close()
+	// isReady returns if the implementation is ready to dial
+	ready() <-chan error
 }
 
 // nopCloser is a mixin to implement a do-nothing close() method of proxyImpl.
@@ -250,7 +252,8 @@ func createImpl(configDir, name, addr, transport string, s *config.ProxyConfig,
 func isAMultiplexedTransport(transport string) bool {
 	return transport == "tlsmasq" ||
 		transport == "starbridge" ||
-		transport == "algeneva"
+		transport == "algeneva" ||
+		transport == "water"
 }
 
 // ForceProxy forces everything through the HTTP proxy at forceAddr using
@@ -339,7 +342,7 @@ func newProxy(name, addr, protocol, network string, s *config.ProxyConfig, uc co
 		name:             name,
 		protocol:         protocol,
 		network:          network,
-		multiplexed:      s.MultiplexedAddr != "",
+		multiplexed:      s.MultiplexedAddr != "" || isAMultiplexedTransport(name),
 		addr:             addr,
 		authToken:        s.AuthToken,
 		user:             uc,
@@ -583,3 +586,7 @@ func splitClientHello(hello []byte) [][]byte {
 	splits = append(splits, hello[start:])
 	return splits
 }
+
+func (p *proxy) Ready() <-chan error {
+	return p.impl.ready()
+}

chained/quic_impl.go

@@ -71,3 +71,7 @@ func (impl *quicImpl) dialServer(op *ops.Op, ctx context.Context) (net.Conn, err
 		return conn, err
 	})
 }
+
+func (*quicImpl) ready() <-chan error {
+	return nil
+}

chained/shadowsocks_impl.go

@@ -36,6 +36,7 @@ type shadowsocksImpl struct {
 	rng            *mrand.Rand
 	rngmx          sync.Mutex
 	tlsConfig      *tls.Config
+	nopCloser
 }
 
 type PrefixSaltGen struct {
@@ -125,9 +126,6 @@ func newShadowsocksImpl(name, addr string, pc *config.ProxyConfig, reportDialCor
 	}, nil
 }
 
-func (impl *shadowsocksImpl) close() {
-}
-
 func (impl *shadowsocksImpl) dialServer(op *ops.Op, ctx context.Context) (net.Conn, error) {
 	return impl.reportDialCore(op, func() (net.Conn, error) {
 		conn, err := impl.client.DialStream(ctx, impl.generateUpstream())
@@ -142,6 +140,10 @@ func (impl *shadowsocksImpl) dialServer(op *ops.Op, ctx context.Context) (net.Co
 	})
 }
 
+func (*shadowsocksImpl) ready() <-chan error {
+	return nil
+}
+
 // generateUpstream() creates a marker upstream address.  This isn't an
 // acutal upstream that will be dialed, it signals that the upstream
 // should be determined by other methods.  It's just a bit random just to

chained/starbridge_impl.go

@@ -67,6 +67,10 @@ func newStarbridgeImpl(name, addr string, pc *config.ProxyConfig, reportDialCore
 	}, nil
 }
 
+func (*starbridge) ready() <-chan error {
+	return nil
+}
+
 // Adapted from https://github.com/OperatorFoundation/Starbridge-go/blob/v3.0.12/Starbridge/v3/starbridge.go#L237-L253
 func getClientConfig(serverPublicKey string) replicant.ClientConfig {
 	polishClientConfig := polish.DarkStarPolishClientConfig{

chained/tlsmasq_impl.go

@@ -111,6 +111,10 @@ func newTLSMasqImpl(configDir, name, addr string, pc *config.ProxyConfig, uc com
 	}, nil
 }
 
+func (*tlsMasqImpl) ready() <-chan error {
+	return nil
+}
+
 func decodeUint16(s string) (uint16, error) {
 	b, err := hex.DecodeString(strings.TrimPrefix(s, "0x"))
 	if err != nil {

chained/water_impl.go

@@ -23,85 +23,110 @@ type waterImpl struct {
 	raddr          string
 	reportDialCore reportDialCoreFn
 	dialer         water.Dialer
-	wgDownload     *sync.WaitGroup
-	downloadErr    error
-	nopCloser
+	readyChan      chan error
 }
 
-var httpClient *http.Client
+var (
+	// waterHTTPClient is a shared HTTP client for downloading WebAssembly modules.
+	waterHTTPClient *http.Client
+
+	// waterWASMLocks is a map of mutexes protecting RW access to the WebAssembly module files for each
+	// WATER transport. A goroutine must acquire the lock for transport "foo" in order to write the
+	// foo.wasm file or read the foo.wasm file on disk.
+	//
+	// Invariant: entries to this map are never overwritten.
+	waterWASMLocks = make(map[string]*sync.Mutex)
+	// wlLock protects access to the waterWASMLocks map.
+	wlLock = new(sync.Mutex)
+)
 
 func newWaterImpl(dir, addr string, pc *config.ProxyConfig, reportDialCore reportDialCoreFn) (*waterImpl, error) {
 	ctx := context.Background()
 	wasmAvailableAt := ptSetting(pc, "water_available_at")
 	transport := ptSetting(pc, "water_transport")
-	wg := new(sync.WaitGroup)
 	d := &waterImpl{
 		raddr:          addr,
 		reportDialCore: reportDialCore,
-		wgDownload:     wg,
+		readyChan:      make(chan error),
 	}
 
 	b64WASM := ptSetting(pc, "water_wasm")
 	if b64WASM != "" {
-		var err error
-		wasm, err := base64.StdEncoding.DecodeString(b64WASM)
-		if err != nil {
-			return nil, fmt.Errorf("failed to decode water wasm: %w", err)
-		}
+		go func() {
+			wasm, err := base64.StdEncoding.DecodeString(b64WASM)
+			if err != nil {
+				d.readyChan <- log.Errorf("failed to decode water wasm: %w", err)
+				return
+			}
 
-		dialer, err := createDialer(ctx, wasm, transport)
-		if err != nil {
-			return nil, log.Errorf("failed to create dialer: %w", err)
-		}
-		d.dialer = dialer
+			d.dialer, err = createDialer(ctx, wasm, transport)
+			if err != nil {
+				d.readyChan <- log.Errorf("failed to create dialer: %w", err)
+				return
+			}
+			d.readyChan <- nil
+		}()
 		return d, nil
 	}
 
 	if wasmAvailableAt != "" {
-		wg.Add(1)
 		go func() {
-			defer wg.Done()
-			log.Debugf("Loading WASM for %q. If not available, it should try to download from the following URLs: %+v. The file should be available here: %s", transport, strings.Split(wasmAvailableAt, ","), dir)
-			vc := newWaterVersionControl(dir)
-			cli := httpClient
-			if cli == nil {
-				cli = proxied.ChainedThenDirectThenFrontedClient(1*time.Minute, "")
-			}
-			downloader, err := newWaterWASMDownloader(strings.Split(wasmAvailableAt, ","), cli)
-			if err != nil {
-				d.downloadErr = log.Errorf("failed to create wasm downloader: %w", err)
-				return
-			}
+			log.Debugf("Loading WASM for %q. If not available, it should try to download from the following URLs: %+v. The file should be available at: %q", transport, strings.Split(wasmAvailableAt, ","), dir)
 
-			r, err := vc.GetWASM(ctx, transport, downloader)
+			r, err := d.loadWASM(ctx, transport, dir, wasmAvailableAt)
 			if err != nil {
-				d.downloadErr = log.Errorf("failed to get wasm: %w", err)
+				d.readyChan <- log.Errorf("failed to read wasm: %w", err)
 				return
 			}
 			defer r.Close()
-
 			b, err := io.ReadAll(r)
 			if err != nil {
-				d.downloadErr = log.Errorf("failed to read wasm: %w", err)
+				d.readyChan <- log.Errorf("failed to load wasm bytes: %w", err)
 				return
 			}
 
-			if len(b) == 0 {
-				d.downloadErr = log.Errorf("received empty wasm")
-				return
-			}
+			log.Debugf("received wasm with %d bytes", len(b))
 
-			dialer, err := createDialer(ctx, b, transport)
+			d.dialer, err = createDialer(ctx, b, transport)
 			if err != nil {
-				d.downloadErr = log.Errorf("failed to create dialer: %w", err)
+				d.readyChan <- log.Errorf("failed to create dialer: %w", err)
+				return
 			}
-			d.dialer = dialer
+			d.readyChan <- nil
 		}()
 	}
 
 	return d, nil
 }
 
+func (d *waterImpl) loadWASM(ctx context.Context, transport string, dir string, wasmAvailableAt string) (io.ReadCloser, error) {
+	wlLock.Lock()
+	m, ok := waterWASMLocks[transport]
+	if !ok {
+		m = new(sync.Mutex)
+		waterWASMLocks[transport] = m
+	}
+	wlLock.Unlock()
+
+	m.Lock()
+	defer m.Unlock()
+
+	vc := newWaterVersionControl(dir)
+	cli := waterHTTPClient
+	if cli == nil {
+		cli = proxied.ChainedThenDirectThenFrontedClient(1*time.Minute, "")
+	}
+	downloader, err := newWaterWASMDownloader(strings.Split(wasmAvailableAt, ","), cli)
+	if err != nil {
+		return nil, log.Errorf("failed to create wasm downloader: %w", err)
+	}
+	r, err := vc.GetWASM(ctx, transport, downloader)
+	if err != nil {
+		return nil, log.Errorf("failed to get wasm: %w", err)
+	}
+	return r, nil
+}
+
 func createDialer(ctx context.Context, wasm []byte, transport string) (water.Dialer, error) {
 	cfg := &water.Config{
 		TransportModuleBin: wasm,
@@ -115,11 +140,29 @@ func createDialer(ctx context.Context, wasm []byte, transport string) (water.Dia
 	return dialer, nil
 }
 
+func (d *waterImpl) close() {
+	close(d.readyChan)
+}
+
+func (d *waterImpl) ready() <-chan error {
+	return d.readyChan
+}
+
 func (d *waterImpl) dialServer(op *ops.Op, ctx context.Context) (net.Conn, error) {
 	return d.reportDialCore(op, func() (net.Conn, error) {
-		d.wgDownload.Wait()
-		if d.dialer == nil {
-			return nil, log.Errorf("dialer not available: %w", d.downloadErr)
+		// if dialer is not ready, wait until WASM is downloaded or context timeout
+		select {
+		case err, ok := <-d.readyChan:
+			if !ok {
+				return nil, log.Error("dialer closed")
+			}
+			if err != nil {
+				return nil, log.Errorf("failed to load dialer: %w", err)
+			}
+
+			log.Debug("dialer ready!")
+		case <-ctx.Done():
+			return nil, log.Errorf("context completed while waiting for WASM download: %w", ctx.Err())
 		}
 
 		// TODO: At water 0.7.0 (currently), the library is	hanging onto the dial context