diff --git a/go.mod b/go.mod index e486595..a84b986 100644 --- a/go.mod +++ b/go.mod @@ -5,24 +5,35 @@ go 1.26.3 require ( github.com/luxfi/lattice/v7 v7.1.0 github.com/luxfi/math v1.4.0 + github.com/luxfi/zap v0.7.2 github.com/montanaflynn/stats v0.9.0 github.com/spf13/cobra v1.10.2 github.com/zeebo/blake3 v0.2.4 - golang.org/x/crypto v0.49.0 + golang.org/x/crypto v0.52.0 ) require ( github.com/ALTree/bigfloat v0.2.0 // indirect + github.com/cenkalti/backoff v2.2.1+incompatible // indirect + github.com/cloudflare/circl v1.6.3 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/google/go-cmp v0.7.0 // indirect + github.com/grandcat/zeroconf v1.0.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/kr/text v0.2.0 // indirect + github.com/luxfi/accel v1.1.9 // indirect + github.com/luxfi/crypto v1.19.17 // indirect + github.com/luxfi/mdns v0.1.1 // indirect + github.com/miekg/dns v1.1.72 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/spf13/pflag v1.0.10 // indirect github.com/stretchr/testify v1.11.1 // indirect - github.com/zeebo/assert v1.3.0 // indirect golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90 // indirect - golang.org/x/sys v0.42.0 // indirect + golang.org/x/mod v0.36.0 // indirect + golang.org/x/net v0.55.0 // indirect + golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.45.0 // indirect + golang.org/x/tools v0.45.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index b50adf4..5faeece 100644 --- a/go.sum +++ b/go.sum @@ -1,11 +1,17 @@ github.com/ALTree/bigfloat v0.2.0 h1:AwNzawrpFuw55/YDVlcPw0F0cmmXrmngBHhVrvdXPvM= github.com/ALTree/bigfloat v0.2.0/go.mod h1:+NaH2gLeY6RPBPPQf4aRotPPStg+eXc8f9ZaE4vRfD4= +github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4= +github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM= +github.com/cloudflare/circl v1.6.3 h1:9GPOhQGF9MCYUeXyMYlqTR6a5gTrgR/fBLXvUgtVcg8= +github.com/cloudflare/circl v1.6.3/go.mod h1:2eXP6Qfat4O/Yhh8BznvKnJ+uzEoTQ6jVKJRn81BiS4= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/grandcat/zeroconf v1.0.0 h1:uHhahLBKqwWBV6WZUDAT71044vwOTL+McW0mBJvo6kE= +github.com/grandcat/zeroconf v1.0.0/go.mod h1:lTKmG1zh86XyCoUeIHSA4FJMBwCJiQmGfcP2PdzytEs= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= @@ -14,12 +20,27 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/luxfi/accel v1.1.9 h1:Tsk6gXj2uKE19501bD0ajRYdeCHIlTGb6jYyLc+F8hc= +github.com/luxfi/accel v1.1.9/go.mod h1:K00BcnLzEYMPHwCFq8Tf/450ApmTs9xBVvYOnofJCkc= +github.com/luxfi/crypto v1.19.17 h1:l2LLu7UFyICtJVfraLDLRi+lFGiDXKHSL18M9/m1gsQ= +github.com/luxfi/crypto v1.19.17/go.mod h1:INjdZtke85k8hX/QAmTMAY8bbZ4gzGZQLqURg3xf6Gk= github.com/luxfi/lattice/v7 v7.1.0 h1:mr3HvN6olNTS2LT/xAW/JBhTqfvpsGmsopDMeR7BSJs= github.com/luxfi/lattice/v7 v7.1.0/go.mod h1:IaaUN+3ysnBG4BA8ILRYG0j80+qtYDP4C5lkaDb2pDE= github.com/luxfi/math v1.4.0 h1:/sb7Grw3hfO+5INWAWdB95jTvCeXg8fSQxsxDzcFtd4= github.com/luxfi/math v1.4.0/go.mod h1:iW0FOCC8qF2mPE+MakG780CAHA83848lb1L04thA1Pg= +github.com/luxfi/mdns v0.1.1 h1:g2eRr9AXcziPkkcd24M+Qu9ApEpoKKjfI79QSNqv0rQ= +github.com/luxfi/mdns v0.1.1/go.mod h1:dbp5f3h3aE7CGzwbaWzBM9cwdcekhmSrWhQevgYhhNA= +github.com/luxfi/pq v1.0.3 h1:pFlQm1+5FuKTDUh2y/23bXWkN4I2Rc5iuxJypwDFFMs= +github.com/luxfi/pq v1.0.3/go.mod h1:8bppZcRElfrVt0n3nYCZW3iX1TvhvzNbdjNdK1irgIE= +github.com/luxfi/zap v0.7.2 h1:YecWTWNE5PPJXL56sLIkzS8b23bprUwZ5lPAQuLUtTE= +github.com/luxfi/zap v0.7.2/go.mod h1:1k+nwT+JW802YzuPAuf7CxMSGr/qxvbGgGwi5k6X9Ok= +github.com/miekg/dns v1.1.27/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM= +github.com/miekg/dns v1.1.72 h1:vhmr+TF2A3tuoGNkLDFK9zi36F2LS+hKTRW0Uf8kbzI= +github.com/miekg/dns v1.1.72/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs= github.com/montanaflynn/stats v0.9.0 h1:tsBJ0RXwph9BmAuFoCmqGv6e8xa0MENQ8m0ptKq29mQ= github.com/montanaflynn/stats v0.9.0/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= @@ -39,12 +60,34 @@ github.com/zeebo/blake3 v0.2.4/go.mod h1:7eeQ6d2iXWRGF6npfaxl2CU+xy2Fjo2gxeyZGCR github.com/zeebo/pcg v1.0.1 h1:lyqfGeWiv4ahac6ttHs+I5hwtH/+1mrhlCtVNQM2kHo= github.com/zeebo/pcg v1.0.1/go.mod h1:09F0S9iiKrwn9rlI5yjLkmrug154/YRW6KnnXVDM/l4= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= -golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4= -golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988= +golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90 h1:jiDhWWeC7jfWqR9c/uplMOqJ0sbNlNWv0UkzE0vX1MA= golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90/go.mod h1:xE1HEv6b+1SCZ5/uscMRjUBKtIxworgEcEi+/n9NQDQ= -golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= -golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= +golang.org/x/mod v0.36.0 h1:JJjpVx6myfUsUdAzZuOSTTmRE0PfZeNWzzvKrP7amb4= +golang.org/x/mod v0.36.0/go.mod h1:moc6ELqsWcOw5Ef3xVprK5ul/MvtVvkIXLziUOICjUQ= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= +golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/tools v0.0.0-20191216052735-49a3e744a425/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= +golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8= +golang.org/x/tools v0.45.0/go.mod h1:LuUGqqaXcXMEFEruIVJVm5mgDD8vww/z/SR1gQ4uE/0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/main.go b/main.go index 612a206..998da4b 100644 --- a/main.go +++ b/main.go @@ -8,6 +8,7 @@ import ( "net" "os" "strconv" + "strings" "sync" "time" @@ -53,17 +54,28 @@ func main() { os.Exit(1) } - // Initialize P2P communication - comm := &networking.P2PComm{ - Socks: make(map[int]*net.Conn), - Rank: partyID, + // Initialize P2P communication over ZAP transport. + // + // Peers and their listen addresses come from the CORONA_PEERS env + // var, formatted as "rank=host:port,rank=host:port,...". The local + // party's own rank is excluded; every other rank in the ceremony + // must be present. The local listen port is derived from the same + // list (peerAddrs[partyID] specifies "host:port" we bind on). + peerAddrs, listenPort, err := parsePeerEnv(os.Getenv("CORONA_PEERS"), partyID) + if err != nil { + log.Fatalf("CORONA_PEERS: %v", err) } + comm, err := networking.NewP2PComm(partyID, listenPort, true /*noDiscovery*/, nil) + if err != nil { + log.Fatalf("NewP2PComm: %v", err) + } + defer comm.Close() + // Suppress unused-import error from prior wiring. + _ = (*net.Conn)(nil) - // Establish connections - var connWg sync.WaitGroup - connWg.Add(1) - go networking.EstablishConnections(&connWg, comm, partyID, sign.K) - connWg.Wait() + if err := comm.EstablishConnectionsZAP(partyID, sign.K, peerAddrs); err != nil { + log.Fatalf("EstablishConnectionsZAP: %v", err) + } var setupDuration, genDuration, signRound1Duration, signRound2PreprocessDuration, signRound2Duration, finalizeDuration, verifyDuration time.Duration var genStart, genEnd, signRound1Start, signRound1End, signRound2Start, signRound2End, combinerReceiveEnd, combinerFinalizeEnd time.Time @@ -240,3 +252,56 @@ func main() { fmt.Printf("Combiner finalize end timestamp: %s\n", combinerFinalizeEnd.Format("15:04:05.000000")) } } + +// parsePeerEnv parses the CORONA_PEERS env var into a per-rank address +// map, and returns the local listen port for partyID. +// +// Format: "0=host:port,1=host:port,...". Every rank in the ceremony +// must appear. The local party's entry is dropped from the returned +// map but its port is harvested as the listen port. +// +// Returns an error on malformed input or when partyID is absent. +func parsePeerEnv(raw string, partyID int) (map[int]string, int, error) { + if strings.TrimSpace(raw) == "" { + return nil, 0, fmt.Errorf("CORONA_PEERS is empty (format: rank=host:port,rank=host:port,...)") + } + peerAddrs := make(map[int]string) + listenPort := 0 + for _, entry := range strings.Split(raw, ",") { + entry = strings.TrimSpace(entry) + if entry == "" { + continue + } + eq := strings.IndexByte(entry, '=') + if eq < 0 { + return nil, 0, fmt.Errorf("CORONA_PEERS entry %q missing '='", entry) + } + rank, err := strconv.Atoi(strings.TrimSpace(entry[:eq])) + if err != nil { + return nil, 0, fmt.Errorf("CORONA_PEERS entry %q: %w", entry, err) + } + addr := strings.TrimSpace(entry[eq+1:]) + if addr == "" { + return nil, 0, fmt.Errorf("CORONA_PEERS entry %q has empty address", entry) + } + if rank == partyID { + // Local listen port from "host:port" — the host is ignored + // (we always listen on all interfaces). + _, portStr, err := net.SplitHostPort(addr) + if err != nil { + return nil, 0, fmt.Errorf("CORONA_PEERS local addr %q: %w", addr, err) + } + p, err := strconv.Atoi(portStr) + if err != nil { + return nil, 0, fmt.Errorf("CORONA_PEERS local port %q: %w", portStr, err) + } + listenPort = p + continue + } + peerAddrs[rank] = addr + } + if listenPort == 0 { + return nil, 0, fmt.Errorf("CORONA_PEERS missing entry for local rank %d", partyID) + } + return peerAddrs, listenPort, nil +} diff --git a/networking/networking.go b/networking/networking.go index dac7b25..bb9356e 100644 --- a/networking/networking.go +++ b/networking/networking.go @@ -1,10 +1,16 @@ +// Copyright (C) 2026, Lux Industries Inc. All rights reserved. +// See the file LICENSE for licensing terms. + package networking import ( "bufio" + "bytes" + "context" "encoding/binary" "fmt" "log" + "log/slog" "net" "sync" "time" @@ -13,408 +19,548 @@ import ( "github.com/luxfi/lattice/v7/utils/structs" ) -type Communicator interface { - Send(dst int, msg []byte) (int, error) - Recv(src int) ([]byte, int, error) - Close() error -} - +// P2PComm is the Corona ceremony communicator. Each party holds one +// P2PComm which it uses to send Vector/Matrix/byte-bag payloads to +// every other party in the round. +// +// Transport is ZAP (luxfi/zap) — every Send* call wraps the lattice- +// serialized bytes in a single ZAP message and ships it via zap.Node; +// every Recv* call dequeues the next ZAP message from the peer's +// receive queue and reads the lattice object from its body. +// +// The lattice serialization (uint64 length + element bytes) inside the +// message body is byte-equal to the pre-migration bare-TCP encoding — +// the protocol bytes Corona participants compute against did not move. type P2PComm struct { + Rank int + + // Socks is retained for backward compatibility with hand-written + // fixtures that wire net.Pipe pairs directly into the comm. New + // code should not touch this map; use the ZAP-backed transport + // (NewP2PComm / Connect / Listen) instead. + // + // When Socks contains a *net.Conn for a peer, Send*/Recv* fall back + // to the legacy in-memory pipe path so existing unit tests keep + // passing without modification. Socks map[int]*net.Conn - Rank int - mu sync.Mutex // Added mutex for safe concurrent access + + mu sync.Mutex + node *zapNode + logger *slog.Logger } +// NewP2PComm creates a Corona communicator backed by a ZAP node bound +// to the given port. Set noDiscovery=true in tests or topology-known +// deployments and supply peer addresses via Connect. +func NewP2PComm(rank int, port int, noDiscovery bool, logger *slog.Logger) (*P2PComm, error) { + if logger == nil { + logger = slog.Default() + } + zn := newZapNode(rank, port, noDiscovery, logger) + if err := zn.node.Start(); err != nil { + return nil, fmt.Errorf("corona: start zap node: %w", err) + } + return &P2PComm{ + Rank: rank, + Socks: make(map[int]*net.Conn), + node: zn, + logger: logger, + }, nil +} + +// Connect dials the peer at addr and binds it to the given rank. Use +// when topology is known up-front (the historical EstablishConnections +// path); when mDNS discovery is enabled, peers are picked up +// automatically and Connect is unnecessary. +func (comm *P2PComm) Connect(rank int, addr string) error { + if comm.node == nil { + return fmt.Errorf("corona: P2PComm has no zap node (call NewP2PComm)") + } + // Retry: peers may not have bound their listener yet. + var lastErr error + for i := 0; i < 10; i++ { + if err := comm.node.node.ConnectDirect(addr); err == nil { + comm.node.registerPeer(rank) + return nil + } else { + lastErr = err + time.Sleep(500 * time.Millisecond) + } + } + return fmt.Errorf("corona: connect rank=%d addr=%s: %w", rank, addr, lastErr) +} + +// AcceptPeer marks rank as a known peer that will dial us. mDNS-less +// deployments where the lower-rank side dials and the higher-rank side +// listens (Corona's historical convention) use this to pre-register +// the incoming peer so Send* knows the ZAP NodeID to address. +func (comm *P2PComm) AcceptPeer(rank int) { + if comm.node != nil { + comm.node.registerPeer(rank) + } +} + +// Close releases all transport resources. +func (comm *P2PComm) Close() error { + if comm.node != nil { + comm.node.close() + } + comm.mu.Lock() + defer comm.mu.Unlock() + for _, sock := range comm.Socks { + if sock != nil && *sock != nil { + _ = (*sock).Close() + } + } + return nil +} + +// SetSock is retained for legacy hand-written fixtures. func (comm *P2PComm) SetSock(key int, conn *net.Conn) { comm.mu.Lock() defer comm.mu.Unlock() comm.Socks[key] = conn } +// GetSock is retained for legacy hand-written fixtures. func (comm *P2PComm) GetSock(key int) *net.Conn { comm.mu.Lock() defer comm.mu.Unlock() return comm.Socks[key] } -func (comm *P2PComm) SendBytes(writer *bufio.Writer, dst int, msg []byte) (int, error) { - length := make([]byte, 4) - binary.BigEndian.PutUint32(length, uint32(len(msg))) - - n, err := writer.Write(length) - if err != nil { - return 0, err - } - totalBytesSent := n - - for len(msg) > 0 { - n, err = writer.Write(msg) - if err != nil { - return 0, err - } - totalBytesSent += n - msg = msg[n:] - } - - err = writer.Flush() - if err != nil { - return 0, err - } - - return totalBytesSent, nil -} - -func (comm *P2PComm) Recv(reader *bufio.Reader, src int) ([]byte, int, error) { - lengthBuf := make([]byte, 4) - totalBytesRead := 0 - for totalBytesRead < 4 { - n, err := reader.Read(lengthBuf[totalBytesRead:]) - if err != nil { - return nil, totalBytesRead, err - } - totalBytesRead += n - } - length := binary.BigEndian.Uint32(lengthBuf) - - data := make([]byte, length) - bytesRead := 0 - for bytesRead < int(length) { - n, err := reader.Read(data[bytesRead:]) - if err != nil { - return nil, totalBytesRead, err - } - bytesRead += n - totalBytesRead += n - } - - return data, totalBytesRead, nil -} - -func (comm *P2PComm) Close() error { +// hasLegacySock reports whether the (legacy) Socks map has a live conn +// for the given rank. When true the Send*/Recv* methods route through +// the bufio.Writer/Reader the caller supplies (test compatibility); +// when false they go through the ZAP node. +func (comm *P2PComm) hasLegacySock(rank int) bool { comm.mu.Lock() defer comm.mu.Unlock() - for _, sock := range comm.Socks { - err := (*sock).Close() + sock, ok := comm.Socks[rank] + return ok && sock != nil && *sock != nil +} + +// --------------------------------------------------------------------- +// Send / Recv primitives. +// +// Every method has two paths: +// +// 1. Legacy direct-conn path. If hasLegacySock(dst) is true (i.e. +// a *net.Conn was wired in via SetSock), the call writes/reads +// to the supplied bufio.Writer/Reader directly. This keeps the +// net.Pipe-based unit tests in networking_test.go working without +// modification. +// +// 2. ZAP path. Otherwise the call serializes the lattice object to +// an in-memory bytes.Buffer, ships the bytes as one ZAP message +// to the peer, and on the receive side dequeues the next ZAP +// message from the per-peer FIFO and parses the lattice object +// from its body. +// +// The lattice serialization itself (vector.go/matrix.go WriteTo/ +// ReadFrom in luxfi/lattice) is identical across both paths — the +// "semantic-level wire" the ceremony computes against is byte-equal. +// --------------------------------------------------------------------- + +// SendVector serializes a lattice vector to bytes and ships it to dst. +// writer is honored when present (legacy path); otherwise the ZAP node +// is used. +func (comm *P2PComm) SendVector(writer *bufio.Writer, dst int, msg structs.Vector[ring.Poly]) { + if writer != nil && comm.hasLegacySock(dst) { + if _, err := msg.WriteTo(writer); err != nil { + log.Fatalf("Failed to write vector: %v", err) + } + if err := writer.Flush(); err != nil { + log.Fatalf("Failed to flush writer: %v", err) + } + return + } + var buf bytes.Buffer + bw := bufio.NewWriter(&buf) + if _, err := msg.WriteTo(bw); err != nil { + log.Fatalf("Failed to write vector: %v", err) + } + if err := bw.Flush(); err != nil { + log.Fatalf("Failed to flush vector buffer: %v", err) + } + if err := comm.node.send(context.Background(), dst, buf.Bytes()); err != nil { + log.Fatalf("Failed to send vector to %d: %v", dst, err) + } +} + +// RecvVector reads a lattice vector of the given length from src. +func (comm *P2PComm) RecvVector(reader *bufio.Reader, src int, length int) structs.Vector[ring.Poly] { + if reader != nil && comm.hasLegacySock(src) { + vec := make(structs.Vector[ring.Poly], length) + if _, err := vec.ReadFrom(reader); err != nil { + log.Fatalf("Failed to read vector: %v", err) + } + return vec + } + body, err := comm.node.recvBody(src) + if err != nil { + log.Fatalf("Failed to recv vector from %d: %v", src, err) + } + vec := make(structs.Vector[ring.Poly], length) + if _, err := vec.ReadFrom(bufio.NewReader(bytes.NewReader(body))); err != nil { + log.Fatalf("Failed to read vector: %v", err) + } + return vec +} + +// SendMatrix serializes a lattice matrix to bytes and ships it to dst. +func (comm *P2PComm) SendMatrix(writer *bufio.Writer, dst int, msg structs.Matrix[ring.Poly]) { + if writer != nil && comm.hasLegacySock(dst) { + if _, err := msg.WriteTo(writer); err != nil { + log.Fatalf("Error sending matrix: %v", err) + } + if err := writer.Flush(); err != nil { + log.Fatalf("Failed to flush writer: %v", err) + } + return + } + var buf bytes.Buffer + bw := bufio.NewWriter(&buf) + if _, err := msg.WriteTo(bw); err != nil { + log.Fatalf("Error sending matrix: %v", err) + } + if err := bw.Flush(); err != nil { + log.Fatalf("Failed to flush matrix buffer: %v", err) + } + if err := comm.node.send(context.Background(), dst, buf.Bytes()); err != nil { + log.Fatalf("Failed to send matrix to %d: %v", dst, err) + } +} + +// RecvMatrix reads a lattice matrix of the given outer length from src. +func (comm *P2PComm) RecvMatrix(reader *bufio.Reader, src int, length int) structs.Matrix[ring.Poly] { + if reader != nil && comm.hasLegacySock(src) { + matrix := make(structs.Matrix[ring.Poly], length) + if _, err := matrix.ReadFrom(reader); err != nil { + log.Fatalf("Failed to read matrix: %v", err) + } + return matrix + } + body, err := comm.node.recvBody(src) + if err != nil { + log.Fatalf("Failed to recv matrix from %d: %v", src, err) + } + matrix := make(structs.Matrix[ring.Poly], length) + if _, err := matrix.ReadFrom(bufio.NewReader(bytes.NewReader(body))); err != nil { + log.Fatalf("Failed to read matrix: %v", err) + } + return matrix +} + +// SendBytesSlice ships a slice of byte-slices to dst with explicit +// length tags. Wire layout: uint32 numSlices, then for each slice +// uint32 length + slice bytes. +func (comm *P2PComm) SendBytesSlice(writer *bufio.Writer, dst int, data [][]byte) { + body := encodeBytesSlice(data) + if writer != nil && comm.hasLegacySock(dst) { + if _, err := writer.Write(body); err != nil { + log.Fatalf("Failed to write bytes slice: %v", err) + } + if err := writer.Flush(); err != nil { + log.Fatalf("Failed to flush writer: %v", err) + } + return + } + if err := comm.node.send(context.Background(), dst, body); err != nil { + log.Fatalf("Failed to send bytes slice to %d: %v", dst, err) + } +} + +// RecvBytesSlice reads a length-prefixed slice-of-slices from src. +func (comm *P2PComm) RecvBytesSlice(reader *bufio.Reader, src int) [][]byte { + if reader != nil && comm.hasLegacySock(src) { + return decodeBytesSlice(reader) + } + body, err := comm.node.recvBody(src) + if err != nil { + log.Fatalf("Failed to recv bytes slice from %d: %v", src, err) + } + return decodeBytesSlice(bufio.NewReader(bytes.NewReader(body))) +} + +// SendBytesMap ships a map to dst. +func (comm *P2PComm) SendBytesMap(writer *bufio.Writer, dst int, data map[int][]byte) { + body := encodeBytesMap(data) + if writer != nil && comm.hasLegacySock(dst) { + if _, err := writer.Write(body); err != nil { + log.Fatalf("Failed to write bytes map: %v", err) + } + if err := writer.Flush(); err != nil { + log.Fatalf("Failed to flush writer: %v", err) + } + return + } + if err := comm.node.send(context.Background(), dst, body); err != nil { + log.Fatalf("Failed to send bytes map to %d: %v", dst, err) + } +} + +// RecvBytesMap reads a map from src. +func (comm *P2PComm) RecvBytesMap(reader *bufio.Reader, src int) map[int][]byte { + if reader != nil && comm.hasLegacySock(src) { + return decodeBytesMap(reader) + } + body, err := comm.node.recvBody(src) + if err != nil { + log.Fatalf("Failed to recv bytes map from %d: %v", src, err) + } + return decodeBytesMap(bufio.NewReader(bytes.NewReader(body))) +} + +// SendBytesSliceMap ships a map to dst. +func (comm *P2PComm) SendBytesSliceMap(writer *bufio.Writer, dst int, data map[int][][]byte) { + body := encodeBytesSliceMap(data) + if writer != nil && comm.hasLegacySock(dst) { + if _, err := writer.Write(body); err != nil { + log.Fatalf("Failed to write bytes-slice map: %v", err) + } + if err := writer.Flush(); err != nil { + log.Fatalf("Failed to flush writer: %v", err) + } + return + } + if err := comm.node.send(context.Background(), dst, body); err != nil { + log.Fatalf("Failed to send bytes-slice map to %d: %v", dst, err) + } +} + +// RecvBytesSliceMap reads a map from src. +func (comm *P2PComm) RecvBytesSliceMap(reader *bufio.Reader, src int) map[int][][]byte { + if reader != nil && comm.hasLegacySock(src) { + return decodeBytesSliceMap(reader) + } + body, err := comm.node.recvBody(src) + if err != nil { + log.Fatalf("Failed to recv bytes-slice map from %d: %v", src, err) + } + return decodeBytesSliceMap(bufio.NewReader(bytes.NewReader(body))) +} + +// --------------------------------------------------------------------- +// Byte-bag wire format. +// +// These encoders/decoders preserve the historical wire layout used by +// Corona over bare-TCP: big-endian uint32 counts and lengths followed +// by raw byte runs. They are now used inside ZAP message bodies, so +// the ZAP frame contains a verbatim copy of the legacy bytes. +// --------------------------------------------------------------------- + +func encodeBytesSlice(data [][]byte) []byte { + var buf bytes.Buffer + bw := bufio.NewWriter(&buf) + numSlices := uint32(len(data)) + if err := binary.Write(bw, binary.BigEndian, numSlices); err != nil { + log.Fatalf("Failed to write number of slices: %v", err) + } + for _, slice := range data { + length := uint32(len(slice)) + if err := binary.Write(bw, binary.BigEndian, length); err != nil { + log.Fatalf("Failed to write slice length: %v", err) + } + if _, err := bw.Write(slice); err != nil { + log.Fatalf("Failed to write slice data: %v", err) + } + } + if err := bw.Flush(); err != nil { + log.Fatalf("Failed to flush bytes-slice buffer: %v", err) + } + return buf.Bytes() +} + +func decodeBytesSlice(r *bufio.Reader) [][]byte { + var numSlices uint32 + if err := binary.Read(r, binary.BigEndian, &numSlices); err != nil { + log.Fatalf("Failed to read number of slices: %v", err) + } + data := make([][]byte, numSlices) + for i := uint32(0); i < numSlices; i++ { + var length uint32 + if err := binary.Read(r, binary.BigEndian, &length); err != nil { + log.Fatalf("Failed to read slice length: %v", err) + } + slice := make([]byte, length) + bytesRead := 0 + for bytesRead < int(length) { + n, err := r.Read(slice[bytesRead:]) + if err != nil { + log.Fatalf("Failed to read slice data: %v", err) + } + bytesRead += n + } + data[i] = slice + } + return data +} + +func encodeBytesMap(data map[int][]byte) []byte { + var buf bytes.Buffer + bw := bufio.NewWriter(&buf) + numEntries := uint32(len(data)) + if err := binary.Write(bw, binary.BigEndian, numEntries); err != nil { + log.Fatalf("Failed to write number of map entries: %v", err) + } + for key, value := range data { + if err := binary.Write(bw, binary.BigEndian, int32(key)); err != nil { + log.Fatalf("Failed to write map key: %v", err) + } + length := uint32(len(value)) + if err := binary.Write(bw, binary.BigEndian, length); err != nil { + log.Fatalf("Failed to write value length: %v", err) + } + if _, err := bw.Write(value); err != nil { + log.Fatalf("Failed to write value data: %v", err) + } + } + if err := bw.Flush(); err != nil { + log.Fatalf("Failed to flush bytes-map buffer: %v", err) + } + return buf.Bytes() +} + +func decodeBytesMap(r *bufio.Reader) map[int][]byte { + var numEntries uint32 + if err := binary.Read(r, binary.BigEndian, &numEntries); err != nil { + log.Fatalf("Failed to read number of map entries: %v", err) + } + data := make(map[int][]byte, numEntries) + for i := uint32(0); i < numEntries; i++ { + var key int32 + if err := binary.Read(r, binary.BigEndian, &key); err != nil { + log.Fatalf("Failed to read map key: %v", err) + } + var length uint32 + if err := binary.Read(r, binary.BigEndian, &length); err != nil { + log.Fatalf("Failed to read value length: %v", err) + } + value := make([]byte, length) + bytesRead := 0 + for bytesRead < int(length) { + n, err := r.Read(value[bytesRead:]) + if err != nil { + log.Fatalf("Failed to read value data: %v", err) + } + bytesRead += n + } + data[int(key)] = value + } + return data +} + +func encodeBytesSliceMap(data map[int][][]byte) []byte { + var buf bytes.Buffer + bw := bufio.NewWriter(&buf) + numEntries := uint32(len(data)) + if err := binary.Write(bw, binary.BigEndian, numEntries); err != nil { + log.Fatalf("Failed to write number of map entries: %v", err) + } + for key, value := range data { + if err := binary.Write(bw, binary.BigEndian, int32(key)); err != nil { + log.Fatalf("Failed to write map key: %v", err) + } + numSlices := uint32(len(value)) + if err := binary.Write(bw, binary.BigEndian, numSlices); err != nil { + log.Fatalf("Failed to write number of slices: %v", err) + } + for _, slice := range value { + length := uint32(len(slice)) + if err := binary.Write(bw, binary.BigEndian, length); err != nil { + log.Fatalf("Failed to write slice length: %v", err) + } + if _, err := bw.Write(slice); err != nil { + log.Fatalf("Failed to write slice data: %v", err) + } + } + } + if err := bw.Flush(); err != nil { + log.Fatalf("Failed to flush bytes-slice map buffer: %v", err) + } + return buf.Bytes() +} + +func decodeBytesSliceMap(r *bufio.Reader) map[int][][]byte { + var numEntries uint32 + if err := binary.Read(r, binary.BigEndian, &numEntries); err != nil { + log.Fatalf("Failed to read number of map entries: %v", err) + } + data := make(map[int][][]byte, numEntries) + for i := uint32(0); i < numEntries; i++ { + var key int32 + if err := binary.Read(r, binary.BigEndian, &key); err != nil { + log.Fatalf("Failed to read map key: %v", err) + } + var numSlices uint32 + if err := binary.Read(r, binary.BigEndian, &numSlices); err != nil { + log.Fatalf("Failed to read number of slices: %v", err) + } + slices := make([][]byte, numSlices) + for j := uint32(0); j < numSlices; j++ { + var length uint32 + if err := binary.Read(r, binary.BigEndian, &length); err != nil { + log.Fatalf("Failed to read slice length: %v", err) + } + slice := make([]byte, length) + bytesRead := 0 + for bytesRead < int(length) { + n, err := r.Read(slice[bytesRead:]) + if err != nil { + log.Fatalf("Failed to read slice data: %v", err) + } + bytesRead += n + } + slices[j] = slice + } + data[int(key)] = slices + } + return data +} + +// --------------------------------------------------------------------- +// Topology helpers — the historical EstablishConnections path lived +// here. The new ZAP-backed equivalent takes a peer address map from +// the caller rather than hard-coding EC2 IPs. +// --------------------------------------------------------------------- + +// EstablishConnectionsZAP wires all (partyID, otherID) pairs over the +// ZAP transport. peerAddrs[otherID] must be the "host:port" of the +// peer's listening socket. The lower-rank side dials, the higher-rank +// side listens — the historical Corona convention preserved. +func (comm *P2PComm) EstablishConnectionsZAP(partyID, totalParties int, peerAddrs map[int]string) error { + var wg sync.WaitGroup + errCh := make(chan error, totalParties) + for otherID := 0; otherID < totalParties; otherID++ { + if otherID == partyID { + continue + } + wg.Add(1) + go func(otherID int) { + defer wg.Done() + if partyID < otherID { + // We listen — peer will dial us. Pre-register so + // Send* knows the peer's NodeID. + comm.AcceptPeer(otherID) + } else { + addr, ok := peerAddrs[otherID] + if !ok { + errCh <- fmt.Errorf("corona: no address for peer rank %d", otherID) + return + } + if err := comm.Connect(otherID, addr); err != nil { + errCh <- err + } + } + }(otherID) + } + wg.Wait() + close(errCh) + for err := range errCh { if err != nil { return err } } return nil } - -func (comm *P2PComm) SendVector(writer *bufio.Writer, dst int, msg structs.Vector[ring.Poly]) { - if _, err := msg.WriteTo(writer); err != nil { - log.Fatalf("Failed to write vector: %v", err) - } - - if err := writer.Flush(); err != nil { - log.Fatalf("Failed to flush writer: %v", err) - } -} - -func (comm *P2PComm) RecvVector(reader *bufio.Reader, src int, length int) structs.Vector[ring.Poly] { - vec := make(structs.Vector[ring.Poly], length) - if _, err := vec.ReadFrom(reader); err != nil { - log.Fatalf("Failed to read vector: %v", err) - } - return vec -} - -func (comm *P2PComm) SendMatrix(writer *bufio.Writer, dst int, msg structs.Matrix[ring.Poly]) { - if _, err := msg.WriteTo(writer); err != nil { - log.Fatalf("Error sending matrix: %v", err) - } - - if err := writer.Flush(); err != nil { - log.Fatalf("Failed to flush writer: %v", err) - } -} - -func (comm *P2PComm) RecvMatrix(reader *bufio.Reader, src int, length int) structs.Matrix[ring.Poly] { - matrix := make(structs.Matrix[ring.Poly], length) - if _, err := matrix.ReadFrom(reader); err != nil { - log.Fatalf("Failed to read matrix: %v", err) - } - return matrix -} - -func (comm *P2PComm) SendBytesSlice(writer *bufio.Writer, dst int, data [][]byte) { - numSlices := uint32(len(data)) - if err := binary.Write(writer, binary.BigEndian, numSlices); err != nil { - log.Fatalf("Failed to write number of slices: %v", err) - } - - for _, slice := range data { - length := uint32(len(slice)) - if err := binary.Write(writer, binary.BigEndian, length); err != nil { - log.Fatalf("Failed to write slice length: %v", err) - } - - for len(slice) > 0 { - n, err := writer.Write(slice) - if err != nil { - log.Fatalf("Failed to write slice data: %v", err) - } - slice = slice[n:] - } - } - - if err := writer.Flush(); err != nil { - log.Fatalf("Failed to flush writer: %v", err) - } -} - -func (comm *P2PComm) RecvBytesSlice(reader *bufio.Reader, src int) [][]byte { - var numSlices uint32 - if err := binary.Read(reader, binary.BigEndian, &numSlices); err != nil { - log.Fatalf("Failed to read number of slices: %v", err) - } - - data := make([][]byte, numSlices) - for i := uint32(0); i < numSlices; i++ { - var length uint32 - if err := binary.Read(reader, binary.BigEndian, &length); err != nil { - log.Fatalf("Failed to read slice length: %v", err) - } - - slice := make([]byte, length) - bytesRead := 0 - for bytesRead < int(length) { - n, err := reader.Read(slice[bytesRead:]) - if err != nil { - log.Fatalf("Failed to read slice data: %v", err) - } - bytesRead += n - } - - data[i] = slice - } - - return data -} - -func (comm *P2PComm) SendBytesMap(writer *bufio.Writer, dst int, data map[int][]byte) { - numEntries := uint32(len(data)) - if err := binary.Write(writer, binary.BigEndian, numEntries); err != nil { - log.Fatalf("Failed to write number of map entries: %v", err) - } - - for key, value := range data { - if err := binary.Write(writer, binary.BigEndian, int32(key)); err != nil { - log.Fatalf("Failed to write map key: %v", err) - } - - length := uint32(len(value)) - if err := binary.Write(writer, binary.BigEndian, length); err != nil { - log.Fatalf("Failed to write value length: %v", err) - } - - for len(value) > 0 { - n, err := writer.Write(value) - if err != nil { - log.Fatalf("Failed to write value data: %v", err) - } - value = value[n:] - } - } - - if err := writer.Flush(); err != nil { - log.Fatalf("Failed to flush writer: %v", err) - } -} - -func (comm *P2PComm) RecvBytesMap(reader *bufio.Reader, src int) map[int][]byte { - var numEntries uint32 - if err := binary.Read(reader, binary.BigEndian, &numEntries); err != nil { - log.Fatalf("Failed to read number of map entries: %v", err) - } - - data := make(map[int][]byte, numEntries) - for i := uint32(0); i < numEntries; i++ { - var key int32 - if err := binary.Read(reader, binary.BigEndian, &key); err != nil { - log.Fatalf("Failed to read map key: %v", err) - } - - var length uint32 - if err := binary.Read(reader, binary.BigEndian, &length); err != nil { - log.Fatalf("Failed to read value length: %v", err) - } - - value := make([]byte, length) - bytesRead := 0 - for bytesRead < int(length) { - n, err := reader.Read(value[bytesRead:]) - if err != nil { - log.Fatalf("Failed to read value data: %v", err) - } - bytesRead += n - } - - data[int(key)] = value - } - - return data -} - -func (comm *P2PComm) SendBytesSliceMap(writer *bufio.Writer, dst int, data map[int][][]byte) { - numEntries := uint32(len(data)) - if err := binary.Write(writer, binary.BigEndian, numEntries); err != nil { - log.Fatalf("Failed to write number of map entries: %v", err) - } - - for key, value := range data { - if err := binary.Write(writer, binary.BigEndian, int32(key)); err != nil { - log.Fatalf("Failed to write map key: %v", err) - } - - numSlices := uint32(len(value)) - if err := binary.Write(writer, binary.BigEndian, numSlices); err != nil { - log.Fatalf("Failed to write number of slices: %v", err) - } - - for _, slice := range value { - length := uint32(len(slice)) - if err := binary.Write(writer, binary.BigEndian, length); err != nil { - log.Fatalf("Failed to write slice length: %v", err) - } - - for len(slice) > 0 { - n, err := writer.Write(slice) - if err != nil { - log.Fatalf("Failed to write slice data: %v", err) - } - slice = slice[n:] - } - } - } - - if err := writer.Flush(); err != nil { - log.Fatalf("Failed to flush writer: %v", err) - } -} - -func (comm *P2PComm) RecvBytesSliceMap(reader *bufio.Reader, src int) map[int][][]byte { - var numEntries uint32 - if err := binary.Read(reader, binary.BigEndian, &numEntries); err != nil { - log.Fatalf("Failed to read number of map entries: %v", err) - } - - data := make(map[int][][]byte, numEntries) - for i := uint32(0); i < numEntries; i++ { - var key int32 - if err := binary.Read(reader, binary.BigEndian, &key); err != nil { - log.Fatalf("Failed to read map key: %v", err) - } - - var numSlices uint32 - if err := binary.Read(reader, binary.BigEndian, &numSlices); err != nil { - log.Fatalf("Failed to read number of slices: %v", err) - } - - slices := make([][]byte, numSlices) - for j := uint32(0); j < numSlices; j++ { - var length uint32 - if err := binary.Read(reader, binary.BigEndian, &length); err != nil { - log.Fatalf("Failed to read slice length: %v", err) - } - - slice := make([]byte, length) - bytesRead := 0 - for bytesRead < int(length) { - n, err := reader.Read(slice[bytesRead:]) - if err != nil { - log.Fatalf("Failed to read slice data: %v", err) - } - bytesRead += n - } - - slices[j] = slice - } - - data[int(key)] = slices - } - - return data -} - -func ListenTCP(comm *P2PComm, port string, src int) { - l, err := net.Listen("tcp", "0.0.0.0:"+port) - if err != nil { - log.Fatal(err) - return - } - defer l.Close() - for { - conn, err := l.Accept() - if err != nil { - log.Println(err) - continue - } - - comm.SetSock(src, &conn) - break - } -} - -func DialTCP(comm *P2PComm, dst int, address string) { - log.Println("Trying to dial", dst, "at address", address) - - // Check if the socket for the destination is already initialized - sock := comm.GetSock(dst) - if sock != nil && *sock != nil { - log.Printf("Socket for destination %d is already initialized.", dst) - return - } - - var conn net.Conn - var err error - - for i := 0; i < 10; i++ { - conn, err = net.Dial("tcp", address) - if err == nil { - break - } - log.Printf("Failed to dial TCP to %s: %v, retrying...", address, err) - time.Sleep(1 * time.Second) - } - - if err != nil { - log.Fatalf("Failed to dial TCP to %s after retries: %v", address, err) - return - } - - comm.SetSock(dst, &conn) - log.Println("Successfully connected to", address) -} - -func EstablishConnections(wg *sync.WaitGroup, comm *P2PComm, partyID int, totalParties int) { - defer wg.Done() - var localWg sync.WaitGroup - - // List of IP addresses of all parties - ips := []string{"50.18.79.144", "34.226.247.5", "18.199.237.28", "34.244.57.20", "54.248.28.234", "47.129.58.89", "54.232.160.71", "3.107.98.51"} - for otherID := 0; otherID < totalParties; otherID++ { - if partyID != otherID { - localWg.Add(1) - go func(otherID int) { - defer localWg.Done() - port := 6000 + calculatePortOffset(partyID, otherID) - address := fmt.Sprintf("%s:%d", ips[otherID], port) - if partyID < otherID { - ListenTCP(comm, fmt.Sprintf("%d", port), otherID) - } else { - DialTCP(comm, otherID, address) - } - }(otherID) - } - } - localWg.Wait() -} - -func calculatePortOffset(partyID, otherID int) int { - if partyID < otherID { - return partyID*100 + otherID - } - return otherID*100 + partyID -} diff --git a/networking/zap.go b/networking/zap.go new file mode 100644 index 0000000..d95ce5b --- /dev/null +++ b/networking/zap.go @@ -0,0 +1,289 @@ +// Copyright (C) 2026, Lux Industries Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +// Package networking provides the Corona ceremony transport layer. +// +// Wire model. Every Corona send (vector, matrix, byte-slice, byte-map, +// byte-slice-map) is encoded to bytes via the lattice WriteTo/ReadFrom +// routines (vector.go and friends in luxfi/lattice/v7/utils/structs). +// Those bytes are byte-equal regardless of transport — that is the +// "semantic-level wire" Corona participants compute against. +// +// Pre-migration the bytes were length-prefixed and shipped raw over a +// bare TCP socket. Post-migration they ride inside a ZAP message body +// over zap.Node, which gives us: +// +// - One framed transport (no manual binary.BigEndian length-prefix +// loops, no zero-copy regressions on every recv). +// - Multi-transport (TCP today, QUIC tomorrow with an unchanged +// P2PComm API). +// - mDNS discovery and ConnectDirect for tests and topology-known +// deployments. +// +// One ZAP message == one logical Corona send. The receiver dequeues one +// message and parses the lattice object from its body. The lattice +// length tags inside the body are preserved verbatim. +package networking + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "sync" + + "github.com/luxfi/zap" +) + +// CoronaMsgType is the ZAP message-type slot used by Corona's ceremony +// transport. It lives inside this package as an internal kind-byte +// namespace — no LP allocation needed because the only sender and +// receiver are Corona networking peers talking to each other. +// +// ZAP encodes the message type in the high byte of the 16-bit Flags +// field (msgType == Flags >> 8), so the kind-byte range is 0..255. +// 0xCE reads "Corona Ceremony"; we own this slot inside the +// corona/networking package only. +// +// All Corona traffic uses this single type because the call sequence +// on each side determines round semantics (R1 share, R2 nonce- +// commitment, R3 signature share, abort) — exactly as it did over +// bare TCP. +const CoronaMsgType uint8 = 0xCE + +// CoronaServiceType is the mDNS service tag for Corona ceremony peers. +// Tests and unit fixtures set NodeConfig.NoDiscovery=true and use +// ConnectDirect with an explicit address; production deployments can +// flip this on for zero-config discovery on a LAN. +const CoronaServiceType = "_corona._tcp" + +// peerStream is the receive side of one Corona peer relationship. +// Send goes through zap.Node.Send; receive blocks on a FIFO of message +// bodies pushed by the ZAP handler. +type peerStream struct { + mu sync.Mutex + cond *sync.Cond + queue [][]byte + closed bool + peerID string // ZAP peer ID (string-form of integer party ID) +} + +func newPeerStream() *peerStream { + ps := &peerStream{} + ps.cond = sync.NewCond(&ps.mu) + return ps +} + +// push appends a message body to the queue and wakes one waiter. +func (ps *peerStream) push(body []byte) { + ps.mu.Lock() + if ps.closed { + ps.mu.Unlock() + return + } + ps.queue = append(ps.queue, body) + ps.cond.Signal() + ps.mu.Unlock() +} + +// pop dequeues the next message body, blocking if empty. Returns +// io.EOF if the stream is closed and the queue is drained. +func (ps *peerStream) pop() ([]byte, error) { + ps.mu.Lock() + defer ps.mu.Unlock() + for len(ps.queue) == 0 { + if ps.closed { + return nil, io.EOF + } + ps.cond.Wait() + } + body := ps.queue[0] + ps.queue = ps.queue[1:] + return body, nil +} + +// close stops further pushes and wakes all waiters with io.EOF. +func (ps *peerStream) close() { + ps.mu.Lock() + ps.closed = true + ps.cond.Broadcast() + ps.mu.Unlock() +} + +// zapNode is the runtime backing a P2PComm. One per local party. +type zapNode struct { + node *zap.Node + logger *slog.Logger + streams map[int]*peerStream // dst rank -> recv queue + peers map[int]string // rank -> zap NodeID + mu sync.RWMutex +} + +// peerID encodes an integer party rank as a ZAP NodeID. Stable, +// deterministic, easy to reverse for handler dispatch. +func peerID(rank int) string { + return fmt.Sprintf("corona-%d", rank) +} + +// rankFromPeerID is the inverse of peerID. Returns -1 if the peer +// is not a corona party (e.g. a stray mDNS neighbor on the LAN). +func rankFromPeerID(id string) int { + var rank int + if _, err := fmt.Sscanf(id, "corona-%d", &rank); err != nil { + return -1 + } + return rank +} + +// newZapNode creates a ZAP node bound to the given port. The caller is +// responsible for Start() and Stop(). +func newZapNode(rank int, port int, noDiscovery bool, logger *slog.Logger) *zapNode { + if logger == nil { + logger = slog.Default() + } + zn := &zapNode{ + logger: logger, + streams: make(map[int]*peerStream), + peers: make(map[int]string), + } + zn.node = zap.NewNode(zap.NodeConfig{ + NodeID: peerID(rank), + ServiceType: CoronaServiceType, + Port: port, + NoDiscovery: noDiscovery, + Logger: logger, + }) + zn.node.Handle(uint16(CoronaMsgType), zn.handleIncoming) + return zn +} + +// getOrCreateStream returns the recv queue for src, creating one on +// first use. Caller must hold no locks; the function locks internally. +func (zn *zapNode) getOrCreateStream(src int) *peerStream { + zn.mu.RLock() + ps, ok := zn.streams[src] + zn.mu.RUnlock() + if ok { + return ps + } + zn.mu.Lock() + defer zn.mu.Unlock() + if ps, ok := zn.streams[src]; ok { + return ps + } + ps = newPeerStream() + ps.peerID = peerID(src) + zn.streams[src] = ps + return ps +} + +// handleIncoming is the ZAP handler for Corona ceremony messages. It +// resolves the sender's rank from peerID and pushes the message body +// into that rank's receive queue. +// +// The handler returns (nil, nil): Corona uses one-way sends, not +// request/response correlated calls. The lattice encoding is +// self-delimiting inside the message body. +func (zn *zapNode) handleIncoming(_ context.Context, from string, msg *zap.Message) (*zap.Message, error) { + rank := rankFromPeerID(from) + if rank < 0 { + // Foreign peer — drop silently. Production should not see this + // because the service-type is Corona-specific; tests using + // ConnectDirect always wire the right node IDs. + return nil, nil + } + // Copy the body. The pooled bufRef behind msg.Bytes() will be + // released as soon as the handler returns, but the queue can hold + // the body until any time the receiver dequeues. + body := append([]byte(nil), msg.Bytes()...) + zn.getOrCreateStream(rank).push(body) + return nil, nil +} + +// send wraps body in a ZAP message and ships it to the destination +// rank's peer via zap.Node.Send. The message-type lives in the top +// byte of Flags (Flags >> 8 == CoronaMsgType). +func (zn *zapNode) send(ctx context.Context, dst int, body []byte) error { + zn.mu.RLock() + peer, ok := zn.peers[dst] + zn.mu.RUnlock() + if !ok { + return fmt.Errorf("corona: no peer registered for rank %d", dst) + } + // Build a single-bytes-field ZAP message carrying the lattice + // payload verbatim. Flags top byte = CoronaMsgType for dispatch. + // We use SetBytes (variable-length tail) — the body fits there + // regardless of size, and on the receive side we read it back as + // a Bytes field with the same offset. + b := zap.NewBuilder(len(body) + zap.HeaderSize + 32) + obj := b.StartObject(8) // tail pointer: 4-byte offset + 4-byte length + obj.SetBytes(0, body) + obj.FinishAsRoot() + frame := b.FinishWithFlags(uint16(CoronaMsgType) << 8) + parsed, err := zap.Parse(frame) + if err != nil { + return fmt.Errorf("corona: build zap frame: %w", err) + } + return zn.node.Send(ctx, peer, parsed) +} + +// recvBody dequeues the next message from src and returns the raw +// lattice-serialized body. Blocks until one arrives or the stream +// closes (io.EOF). +// +// One copy: the handler stashed the full ZAP frame in the queue; we +// parse it here and return a slice into that frame. Because the +// queue-owned frame outlives this call (it's freed when GC reclaims +// the queue entry), the returned slice is safe to hold. +func (zn *zapNode) recvBody(src int) ([]byte, error) { + frame, err := zn.getOrCreateStream(src).pop() + if err != nil { + return nil, err + } + msg, err := zap.Parse(frame) + if err != nil { + return nil, fmt.Errorf("corona: parse zap frame: %w", err) + } + body := msg.Root().Bytes(0) + if body == nil { + return nil, errors.New("corona: empty body in ZAP message") + } + return body, nil +} + +// registerPeer binds a party rank to its ZAP peer ID. Called after a +// successful Connect or Listen handshake for the (rank, addr) pair. +func (zn *zapNode) registerPeer(rank int) { + zn.mu.Lock() + zn.peers[rank] = peerID(rank) + zn.mu.Unlock() +} + +// close shuts down the ZAP node and all peer streams. +func (zn *zapNode) close() { + zn.mu.Lock() + for _, ps := range zn.streams { + ps.close() + } + zn.mu.Unlock() + if zn.node != nil { + zn.node.Stop() + } +} + +// CalculatePortOffset is the historical convention used by Corona's +// EstablishConnections to derive a unique port per (partyA, partyB) +// pair. Kept for callers that still drive deployment topology with +// a hard-coded port grid. +// +// CalculatePortOffset(a, b) == CalculatePortOffset(b, a) by design — +// both sides of a pair need to agree on the same port without +// negotiating. +func CalculatePortOffset(partyID, otherID int) int { + if partyID < otherID { + return partyID*100 + otherID + } + return otherID*100 + partyID +} + diff --git a/networking/zap_test.go b/networking/zap_test.go new file mode 100644 index 0000000..6fde383 --- /dev/null +++ b/networking/zap_test.go @@ -0,0 +1,490 @@ +// Copyright (C) 2026, Lux Industries Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package networking + +import ( + "bufio" + "bytes" + "fmt" + "io" + "log/slog" + "net" + "sync" + "testing" + "time" + + "github.com/luxfi/lattice/v7/ring" + "github.com/luxfi/lattice/v7/utils/sampling" + "github.com/luxfi/lattice/v7/utils/structs" +) + +func bufioNewReader(r io.Reader) *bufio.Reader { return bufio.NewReader(r) } +func bufioNewWriter(w io.Writer) *bufio.Writer { return bufio.NewWriter(w) } + +// newSilentLogger returns a slog.Logger that discards everything. Used +// in benchmarks so the per-iteration log overhead does not pollute +// timing numbers. +func newSilentLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError + 1})) +} + +// freePort grabs a free TCP port for the test by binding to :0, +// reading the assigned port, and closing — the kernel may reuse the +// port for the test listener microseconds later. +func freePort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("freePort: %v", err) + } + port := l.Addr().(*net.TCPAddr).Port + _ = l.Close() + return port +} + +// twoPartyZAP brings up two P2PComm instances over ZAP and wires them +// to each other. Returns the two comms plus a cleanup func. +func twoPartyZAP(t *testing.T) (*P2PComm, *P2PComm, func()) { + t.Helper() + port0 := freePort(t) + port1 := freePort(t) + silent := newSilentLogger() + + comm0, err := NewP2PComm(0, port0, true /*noDiscovery*/, silent) + if err != nil { + t.Fatalf("NewP2PComm 0: %v", err) + } + comm1, err := NewP2PComm(1, port1, true /*noDiscovery*/, silent) + if err != nil { + comm0.Close() + t.Fatalf("NewP2PComm 1: %v", err) + } + + // Lower-rank side dials, higher-rank side accepts (Corona's + // historical convention preserved in EstablishConnectionsZAP). + addr1 := fmt.Sprintf("127.0.0.1:%d", port1) + comm0.AcceptPeer(1) // local registers 1's NodeID + if err := comm0.Connect(1, addr1); err != nil { + comm0.Close() + comm1.Close() + t.Fatalf("Connect 0->1: %v", err) + } + comm1.AcceptPeer(0) + + // Allow the listener side a moment to register the inbound conn. + time.Sleep(100 * time.Millisecond) + + cleanup := func() { + comm0.Close() + comm1.Close() + } + return comm0, comm1, cleanup +} + +func TestZAP_SendRecvVector(t *testing.T) { + comm0, comm1, cleanup := twoPartyZAP(t) + defer cleanup() + + r, _ := ring.NewRing(256, []uint64{8380417}) + prng, _ := sampling.NewPRNG() + sampler := ring.NewUniformSampler(prng, r) + testVector := make(structs.Vector[ring.Poly], 3) + for i := range testVector { + testVector[i] = sampler.ReadNew() + } + + var received structs.Vector[ring.Poly] + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + received = comm1.RecvVector(nil, 0, len(testVector)) + }() + + comm0.SendVector(nil, 1, testVector) + + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("ZAP vector recv timed out") + } + + if len(received) != len(testVector) { + t.Fatalf("len: got %d want %d", len(received), len(testVector)) + } + for i := range testVector { + if !r.Equal(received[i], testVector[i]) { + t.Errorf("vec[%d] mismatch", i) + } + } +} + +func TestZAP_SendRecvMatrix(t *testing.T) { + comm0, comm1, cleanup := twoPartyZAP(t) + defer cleanup() + + r, _ := ring.NewRing(256, []uint64{8380417}) + prng, _ := sampling.NewPRNG() + sampler := ring.NewUniformSampler(prng, r) + testMatrix := make(structs.Matrix[ring.Poly], 2) + for i := range testMatrix { + testMatrix[i] = make(structs.Vector[ring.Poly], 3) + for j := range testMatrix[i] { + testMatrix[i][j] = sampler.ReadNew() + } + } + + var received structs.Matrix[ring.Poly] + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + received = comm1.RecvMatrix(nil, 0, len(testMatrix)) + }() + + comm0.SendMatrix(nil, 1, testMatrix) + + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("ZAP matrix recv timed out") + } + + if len(received) != len(testMatrix) { + t.Fatalf("rows: got %d want %d", len(received), len(testMatrix)) + } + for i := range testMatrix { + if len(received[i]) != len(testMatrix[i]) { + t.Errorf("row %d cols: got %d want %d", i, len(received[i]), len(testMatrix[i])) + } + for j := range testMatrix[i] { + if !r.Equal(received[i][j], testMatrix[i][j]) { + t.Errorf("matrix[%d][%d] mismatch", i, j) + } + } + } +} + +func TestZAP_SendRecvBytesSlice(t *testing.T) { + comm0, comm1, cleanup := twoPartyZAP(t) + defer cleanup() + + in := [][]byte{ + []byte("round 1 share"), + []byte("round 2 nonce-commitment"), + []byte("round 3 signature share"), + } + + var out [][]byte + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + out = comm1.RecvBytesSlice(nil, 0) + }() + comm0.SendBytesSlice(nil, 1, in) + + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("ZAP bytes-slice recv timed out") + } + + if len(out) != len(in) { + t.Fatalf("len: got %d want %d", len(out), len(in)) + } + for i := range in { + if !bytes.Equal(out[i], in[i]) { + t.Errorf("slice[%d]: got %q want %q", i, out[i], in[i]) + } + } +} + +func TestZAP_SendRecvBytesMap(t *testing.T) { + comm0, comm1, cleanup := twoPartyZAP(t) + defer cleanup() + + in := map[int][]byte{ + 0: []byte("party-0-share"), + 1: []byte("party-1-share"), + 2: []byte("party-2-share"), + } + + var out map[int][]byte + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + out = comm1.RecvBytesMap(nil, 0) + }() + comm0.SendBytesMap(nil, 1, in) + + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("ZAP bytes-map recv timed out") + } + + if len(out) != len(in) { + t.Fatalf("len: got %d want %d", len(out), len(in)) + } + for k, v := range in { + got, ok := out[k] + if !ok { + t.Errorf("missing key %d", k) + continue + } + if !bytes.Equal(got, v) { + t.Errorf("key %d: got %q want %q", k, got, v) + } + } +} + +func TestZAP_SendRecvBytesSliceMap(t *testing.T) { + comm0, comm1, cleanup := twoPartyZAP(t) + defer cleanup() + + in := map[int][][]byte{ + 0: {[]byte("seed-0-a"), []byte("seed-0-b")}, + 1: {[]byte("seed-1-a"), []byte("seed-1-b"), []byte("seed-1-c")}, + } + + var out map[int][][]byte + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + out = comm1.RecvBytesSliceMap(nil, 0) + }() + comm0.SendBytesSliceMap(nil, 1, in) + + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("ZAP bytes-slice-map recv timed out") + } + + if len(out) != len(in) { + t.Fatalf("len: got %d want %d", len(out), len(in)) + } + for k, vs := range in { + gotVs, ok := out[k] + if !ok { + t.Errorf("missing key %d", k) + continue + } + if len(gotVs) != len(vs) { + t.Errorf("key %d: len got %d want %d", k, len(gotVs), len(vs)) + continue + } + for i := range vs { + if !bytes.Equal(gotVs[i], vs[i]) { + t.Errorf("key %d slice %d: got %q want %q", k, i, gotVs[i], vs[i]) + } + } + } +} + +// TestZAP_RoundtripPreservesLatticeBytes asserts the byte-equal-on- +// semantic-wire invariant: the bytes Corona participants compute +// against (the lattice serialization) are identical to what a direct +// in-memory WriteTo/ReadFrom would produce. The ZAP framing is purely +// envelope. +func TestZAP_RoundtripPreservesLatticeBytes(t *testing.T) { + r, _ := ring.NewRing(256, []uint64{8380417}) + prng, _ := sampling.NewPRNG() + sampler := ring.NewUniformSampler(prng, r) + v := make(structs.Vector[ring.Poly], 4) + for i := range v { + v[i] = sampler.ReadNew() + } + + // Expected bytes: pure lattice serialization. + var expected bytes.Buffer + if _, err := v.WriteTo(&expected); err != nil { + t.Fatalf("expected WriteTo: %v", err) + } + + comm0, comm1, cleanup := twoPartyZAP(t) + defer cleanup() + + var received structs.Vector[ring.Poly] + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + received = comm1.RecvVector(nil, 0, len(v)) + }() + comm0.SendVector(nil, 1, v) + + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("ZAP roundtrip recv timed out") + } + + // The received vector must serialize back to byte-equal output. + var actual bytes.Buffer + if _, err := received.WriteTo(&actual); err != nil { + t.Fatalf("actual WriteTo: %v", err) + } + if !bytes.Equal(expected.Bytes(), actual.Bytes()) { + t.Fatalf("lattice serialization differs across ZAP transport (len exp=%d got=%d)", expected.Len(), actual.Len()) + } +} + +// BenchmarkSendRecvVector_ZAP measures the round-trip cost of the ZAP +// transport for a representative Corona payload. +func BenchmarkSendRecvVector_ZAP(b *testing.B) { + port0, port1 := pickPort(b), pickPort(b) + silent := newSilentLogger() + comm0, err := NewP2PComm(0, port0, true, silent) + if err != nil { + b.Fatalf("NewP2PComm 0: %v", err) + } + defer comm0.Close() + comm1, err := NewP2PComm(1, port1, true, silent) + if err != nil { + b.Fatalf("NewP2PComm 1: %v", err) + } + defer comm1.Close() + if err := comm0.Connect(1, fmt.Sprintf("127.0.0.1:%d", port1)); err != nil { + b.Fatalf("connect: %v", err) + } + comm1.AcceptPeer(0) + time.Sleep(100 * time.Millisecond) + + r, _ := ring.NewRing(256, []uint64{8380417}) + prng, _ := sampling.NewPRNG() + sampler := ring.NewUniformSampler(prng, r) + v := make(structs.Vector[ring.Poly], 8) + for i := range v { + v[i] = sampler.ReadNew() + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + done := make(chan struct{}) + go func() { + _ = comm1.RecvVector(nil, 0, len(v)) + close(done) + }() + comm0.SendVector(nil, 1, v) + <-done + } +} + +func pickPort(b *testing.B) int { + b.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + b.Fatalf("pickPort: %v", err) + } + p := l.Addr().(*net.TCPAddr).Port + _ = l.Close() + return p +} + +// BenchmarkSendRecvVector_BareTCPPipe is the cheap synchronous in- +// memory baseline. net.Pipe has no TCP-stack overhead — it just hands +// the bytes between goroutines through a mutex. Useful as a floor on +// the lattice-serialization cost alone. +func BenchmarkSendRecvVector_BareTCPPipe(b *testing.B) { + server, client := net.Pipe() + defer server.Close() + defer client.Close() + + comm0 := &P2PComm{Rank: 0, Socks: map[int]*net.Conn{1: &client}} + comm1 := &P2PComm{Rank: 1, Socks: map[int]*net.Conn{0: &server}} + + r, _ := ring.NewRing(256, []uint64{8380417}) + prng, _ := sampling.NewPRNG() + sampler := ring.NewUniformSampler(prng, r) + v := make(structs.Vector[ring.Poly], 8) + for i := range v { + v[i] = sampler.ReadNew() + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + done := make(chan struct{}) + go func() { + reader := bufioNewReader(server) + _ = comm1.RecvVector(reader, 0, len(v)) + close(done) + }() + writer := bufioNewWriter(client) + comm0.SendVector(writer, 1, v) + <-done + } +} + +// BenchmarkSendRecvVector_BareTCPLoopback is the apples-to-apples +// baseline for the ZAP bench: real TCP on the loopback interface +// using the pre-migration manual-length-prefix code path +// (SendVector via writer.Flush over a real net.TCPConn). +func BenchmarkSendRecvVector_BareTCPLoopback(b *testing.B) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + b.Fatalf("listen: %v", err) + } + defer ln.Close() + addr := ln.Addr().String() + + connCh := make(chan net.Conn, 1) + go func() { + c, err := ln.Accept() + if err != nil { + return + } + connCh <- c + }() + clientConn, err := net.Dial("tcp", addr) + if err != nil { + b.Fatalf("dial: %v", err) + } + defer clientConn.Close() + serverConn := <-connCh + defer serverConn.Close() + + var client net.Conn = clientConn + var server net.Conn = serverConn + + comm0 := &P2PComm{Rank: 0, Socks: map[int]*net.Conn{1: &client}} + comm1 := &P2PComm{Rank: 1, Socks: map[int]*net.Conn{0: &server}} + + r, _ := ring.NewRing(256, []uint64{8380417}) + prng, _ := sampling.NewPRNG() + sampler := ring.NewUniformSampler(prng, r) + v := make(structs.Vector[ring.Poly], 8) + for i := range v { + v[i] = sampler.ReadNew() + } + + reader := bufioNewReader(server) + writer := bufioNewWriter(client) + + b.ResetTimer() + for i := 0; i < b.N; i++ { + done := make(chan struct{}) + go func() { + _ = comm1.RecvVector(reader, 0, len(v)) + close(done) + }() + comm0.SendVector(writer, 1, v) + <-done + } +}