network: stateless vote compression (algorand#6276)

Co-authored-by: Copilot <[email protected]>

Author: cce

agreement/vote.go

@@ -146,6 +146,12 @@ func (uv unauthenticatedVote) verify(l LedgerReader) (vote, error) {
 	return vote{R: rv, Cred: cred, Sig: uv.Sig}, nil
 }
 
+var (
+	// testMakeVoteCheck is a function that can be set to check every
+	// unauthenticatedVote before it is returned by makeVote. It is only set by tests.
+	testMakeVoteCheck func(*unauthenticatedVote) error
+)
+
 // makeVote creates a new unauthenticated vote from its constituent components.
 //
 // makeVote returns an error if it fails.
@@ -178,7 +184,15 @@ func makeVote(rv rawVote, voting crypto.OneTimeSigner, selection *crypto.VRFSecr
 	}
 
 	cred := committee.MakeCredential(&selection.SK, m.Selector)
-	return unauthenticatedVote{R: rv, Cred: cred, Sig: sig}, nil
+	ret := unauthenticatedVote{R: rv, Cred: cred, Sig: sig}
+
+	// for use when running in tests
+	if testMakeVoteCheck != nil {
+		if testErr := testMakeVoteCheck(&ret); testErr != nil {
+			return unauthenticatedVote{}, fmt.Errorf("makeVote: testMakeVoteCheck failed: %w", testErr)
+		}
+	}
+	return ret, nil
 }
 
 // ToBeHashed implements the Hashable interface.

agreement/vote_test.go

@@ -17,6 +17,10 @@
 package agreement
 
 import (
+	"bytes"
+	"encoding/base64"
+	"encoding/hex"
+	"fmt"
 	"os"
 	"testing"
 
@@ -27,10 +31,38 @@ import (
 	"github.com/algorand/go-algorand/data/basics"
 	"github.com/algorand/go-algorand/data/committee"
 	"github.com/algorand/go-algorand/logging"
+	"github.com/algorand/go-algorand/network/vpack"
 	"github.com/algorand/go-algorand/protocol"
 	"github.com/algorand/go-algorand/test/partitiontest"
 )
 
+func init() {
+	testMakeVoteCheck = testVPackMakeVote
+}
+
+func testVPackMakeVote(v *unauthenticatedVote) error {
+	vbuf := protocol.Encode(v)
+	enc := vpack.NewStatelessEncoder()
+	dec := vpack.NewStatelessDecoder()
+	encBuf, err := enc.CompressVote(nil, vbuf)
+	if err != nil {
+		return fmt.Errorf("makeVote: failed to parse vote msgpack: %v", err)
+	}
+	decBuf, err := dec.DecompressVote(nil, encBuf)
+	if err != nil {
+		return fmt.Errorf("makeVote: failed to decompress vote msgpack: %v", err)
+	}
+	if !bytes.Equal(vbuf, decBuf) {
+		fmt.Printf("vote: %+v\n", v)
+		fmt.Printf("oldbuf: %s\n", hex.EncodeToString(vbuf))
+		fmt.Printf("decbuf: %s\n", hex.EncodeToString(decBuf))
+		fmt.Printf("base64 oldbuf: %s\n", base64.StdEncoding.EncodeToString(vbuf))
+		fmt.Printf("base64 decbuf: %s\n", base64.StdEncoding.EncodeToString(decBuf))
+		return fmt.Errorf("makeVote: decompressed vote msgpack does not match original")
+	}
+	return nil
+}
+
 // error is set if this address is not selected
 func makeVoteTesting(addr basics.Address, vrfSecs *crypto.VRFSecrets, otSecs crypto.OneTimeSigner, ledger Ledger, round basics.Round, period period, step step, digest crypto.Digest) (vote, error) {
 	var proposal proposalValue

config/localTemplate.go

@@ -44,7 +44,7 @@ type Local struct {
 	// Version tracks the current version of the defaults so we can migrate old -> new
 	// This is specifically important whenever we decide to change the default value
 	// for an existing parameter. This field tag must be updated any time we add a new version.
-	Version uint32 `version[0]:"0" version[1]:"1" version[2]:"2" version[3]:"3" version[4]:"4" version[5]:"5" version[6]:"6" version[7]:"7" version[8]:"8" version[9]:"9" version[10]:"10" version[11]:"11" version[12]:"12" version[13]:"13" version[14]:"14" version[15]:"15" version[16]:"16" version[17]:"17" version[18]:"18" version[19]:"19" version[20]:"20" version[21]:"21" version[22]:"22" version[23]:"23" version[24]:"24" version[25]:"25" version[26]:"26" version[27]:"27" version[28]:"28" version[29]:"29" version[30]:"30" version[31]:"31" version[32]:"32" version[33]:"33" version[34]:"34" version[35]:"35"`
+	Version uint32 `version[0]:"0" version[1]:"1" version[2]:"2" version[3]:"3" version[4]:"4" version[5]:"5" version[6]:"6" version[7]:"7" version[8]:"8" version[9]:"9" version[10]:"10" version[11]:"11" version[12]:"12" version[13]:"13" version[14]:"14" version[15]:"15" version[16]:"16" version[17]:"17" version[18]:"18" version[19]:"19" version[20]:"20" version[21]:"21" version[22]:"22" version[23]:"23" version[24]:"24" version[25]:"25" version[26]:"26" version[27]:"27" version[28]:"28" version[29]:"29" version[30]:"30" version[31]:"31" version[32]:"32" version[33]:"33" version[34]:"34" version[35]:"35" version[36]:"36"`
 
 	// Archival nodes retain a full copy of the block history. Non-Archival nodes will delete old blocks and only retain what's need to properly validate blockchain messages (the precise number of recent blocks depends on the consensus parameters. Currently the last 1321 blocks are required). This means that non-Archival nodes require significantly less storage than Archival nodes.  If setting this to true for the first time, the existing ledger may need to be deleted to get the historical values stored as the setting only affects current blocks forward. To do this, shutdown the node and delete all .sqlite files within the data/testnet-version directory, except the crash.sqlite file. Restart the node and wait for the node to sync.
 	Archival bool `version[0]:"false"`
@@ -643,6 +643,9 @@ type Local struct {
 	// GoMemLimit provides the Go runtime with a soft memory limit. The default behavior is no limit,
 	// unless the GOMEMLIMIT environment variable is set.
 	GoMemLimit uint64 `version[34]:"0"`
+
+	// EnableVoteCompression controls whether vote compression is enabled for websocket networks
+	EnableVoteCompression bool `version[36]:"true"`
 }
 
 // DNSBootstrapArray returns an array of one or more DNS Bootstrap identifiers

config/local_defaults.go

@@ -20,7 +20,7 @@
 package config
 
 var defaultLocal = Local{
-	Version:                                    35,
+	Version:                                    36,
 	AccountUpdatesStatsInterval:                5000000000,
 	AccountsRebuildSynchronousMode:             1,
 	AgreementIncomingBundlesQueueLength:        15,
@@ -89,6 +89,7 @@ var defaultLocal = Local{
 	EnableTxnEvalTracer:                        false,
 	EnableUsageLog:                             false,
 	EnableVerbosedTransactionSyncLogging:       false,
+	EnableVoteCompression:                      true,
 	EndpointAddress:                            "127.0.0.1:0",
 	FallbackDNSResolverAddress:                 "",
 	ForceFetchTransactions:                     false,

go.mod

@@ -58,7 +58,7 @@ require (
 	golang.org/x/sys v0.30.0
 	golang.org/x/text v0.22.0
 	gopkg.in/sohlich/elogrus.v3 v3.0.0-20180410122755-1fa29e2f2009
-	pgregory.net/rapid v0.6.2
+	pgregory.net/rapid v1.2.0
 )
 
 require (

go.sum

@@ -970,8 +970,8 @@ honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 lukechampine.com/blake3 v1.3.0 h1:sJ3XhFINmHSrYCgl958hscfIa3bw8x4DqMP3u1YvoYE=
 lukechampine.com/blake3 v1.3.0/go.mod h1:0OFRp7fBtAylGVCO40o87sbupkyIGgbpv1+M1k1LM6k=
-pgregory.net/rapid v0.6.2 h1:ErW5sL+UKtfBfUTsWHDCoeB+eZKLKMxrSd1VJY6W4bw=
-pgregory.net/rapid v0.6.2/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=
+pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk=
+pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=
 rsc.io/tmplfunc v0.0.3 h1:53XFQh69AfOa8Tw0Jm7t+GV7KZhOi6jzsCzTtKbMvzU=
 rsc.io/tmplfunc v0.0.3/go.mod h1:AG3sTPzElb1Io3Yg4voV9AGZJuleGAwaVRxL9M49PhA=
 sourcegraph.com/sourcegraph/go-diff v0.5.0/go.mod h1:kuch7UrkMzY0X+p9CRK03kfuPQ2zzQcaEFbx8wA8rck=

installer/config.json.example

@@ -1,5 +1,5 @@
 {
-    "Version": 35,
+    "Version": 36,
     "AccountUpdatesStatsInterval": 5000000000,
     "AccountsRebuildSynchronousMode": 1,
     "AgreementIncomingBundlesQueueLength": 15,
@@ -68,6 +68,7 @@
     "EnableTxnEvalTracer": false,
     "EnableUsageLog": false,
     "EnableVerbosedTransactionSyncLogging": false,
+    "EnableVoteCompression": true,
     "EndpointAddress": "127.0.0.1:0",
     "FallbackDNSResolverAddress": "",
     "ForceFetchTransactions": false,

network/msgCompressor.go

@@ -24,6 +24,7 @@ import (
 	"github.com/DataDog/zstd"
 
 	"github.com/algorand/go-algorand/logging"
+	"github.com/algorand/go-algorand/network/vpack"
 	"github.com/algorand/go-algorand/protocol"
 )
 
@@ -52,19 +53,39 @@ func zstdCompressMsg(tbytes []byte, d []byte) ([]byte, string) {
 	return mbytesComp, ""
 }
 
+func vpackCompressVote(tbytes []byte, d []byte) ([]byte, string) {
+	var enc vpack.StatelessEncoder
+	bound := vpack.MaxCompressedVoteSize
+	// Pre-allocate buffer for tag bytes and compressed data
+	mbytesComp := make([]byte, len(tbytes)+bound)
+	copy(mbytesComp, tbytes)
+	comp, err := enc.CompressVote(mbytesComp[len(tbytes):], d)
+	if err != nil {
+		// fallback and reuse non-compressed original data
+		logMsg := fmt.Sprintf("failed to compress vote into buffer of len %d: %v", len(d), err)
+		copied := copy(mbytesComp[len(tbytes):], d)
+		return mbytesComp[:len(tbytes)+copied], logMsg
+	}
+
+	result := mbytesComp[:len(tbytes)+len(comp)]
+	return result, ""
+}
+
 // MaxDecompressedMessageSize defines a maximum decompressed data size
 // to prevent zip bombs. This depends on MaxTxnBytesPerBlock consensus parameter
 // and should be larger.
 const MaxDecompressedMessageSize = 20 * 1024 * 1024 // some large enough value
 
-// wsPeerMsgDataConverter performs optional incoming messages conversion.
-// At the moment it only supports zstd decompression for payload proposal
-type wsPeerMsgDataConverter struct {
+// wsPeerMsgDataDecoder performs optional incoming messages conversion.
+// At the moment it only supports zstd decompression for payload proposal,
+// and vpack decompression for votes.
+type wsPeerMsgDataDecoder struct {
 	log    logging.Logger
 	origin string
 
 	// actual converter(s)
 	ppdec zstdProposalDecompressor
+	avdec vpackVoteDecompressor
 }
 
 type zstdProposalDecompressor struct{}
@@ -73,6 +94,15 @@ func (dec zstdProposalDecompressor) accept(data []byte) bool {
 	return len(data) > 4 && bytes.Equal(data[:4], zstdCompressionMagic[:])
 }
 
+type vpackVoteDecompressor struct {
+	enabled bool
+	dec     *vpack.StatelessDecoder
+}
+
+func (dec vpackVoteDecompressor) convert(data []byte) ([]byte, error) {
+	return dec.dec.DecompressVote(nil, data)
+}
+
 func (dec zstdProposalDecompressor) convert(data []byte) ([]byte, error) {
 	r := zstd.NewReader(bytes.NewReader(data))
 	defer r.Close()
@@ -96,7 +126,7 @@ func (dec zstdProposalDecompressor) convert(data []byte) ([]byte, error) {
 	}
 }
 
-func (c *wsPeerMsgDataConverter) convert(tag protocol.Tag, data []byte) ([]byte, error) {
+func (c *wsPeerMsgDataDecoder) convert(tag protocol.Tag, data []byte) ([]byte, error) {
 	if tag == protocol.ProposalPayloadTag {
 		// sender might support compressed payload but fail to compress for whatever reason,
 		// in this case it sends non-compressed payload - the receiver decompress only if it is compressed.
@@ -108,16 +138,33 @@ func (c *wsPeerMsgDataConverter) convert(tag protocol.Tag, data []byte) ([]byte,
 			return res, nil
 		}
 		c.log.Warnf("peer %s supported zstd but sent non-compressed data", c.origin)
+	} else if tag == protocol.AgreementVoteTag {
+		if c.avdec.enabled {
+			res, err := c.avdec.convert(data)
+			if err != nil {
+				c.log.Warnf("peer %s vote decompress error: %v", c.origin, err)
+				// fall back to original data
+				return data, nil
+			}
+			return res, nil
+		}
 	}
 	return data, nil
 }
 
-func makeWsPeerMsgDataConverter(wp *wsPeer) *wsPeerMsgDataConverter {
-	c := wsPeerMsgDataConverter{
+func makeWsPeerMsgDataDecoder(wp *wsPeer) *wsPeerMsgDataDecoder {
+	c := wsPeerMsgDataDecoder{
 		log:    wp.log,
 		origin: wp.originAddress,
 	}
 
 	c.ppdec = zstdProposalDecompressor{}
+	// have both ends advertised support for compression?
+	if wp.enableVoteCompression && wp.vpackVoteCompressionSupported() {
+		c.avdec = vpackVoteDecompressor{
+			enabled: true,
+			dec:     vpack.NewStatelessDecoder(),
+		}
+	}
 	return &c
 }

network/msgCompressor_test.go

@@ -76,7 +76,7 @@ func (cl *converterTestLogger) Warnf(s string, args ...interface{}) {
 func TestWsPeerMsgDataConverterConvert(t *testing.T) {
 	partitiontest.PartitionTest(t)
 
-	c := wsPeerMsgDataConverter{}
+	c := wsPeerMsgDataDecoder{}
 	c.ppdec = zstdProposalDecompressor{}
 	tag := protocol.AgreementVoteTag
 	data := []byte("data")

network/vpack/README.md

@@ -0,0 +1,72 @@
+# Stateless *vpack* wire format
+
+This document specifies the byte‑level (on‑wire) layout produced by `StatelessEncoder.CompressVote` and accepted by `StatelessDecoder.DecompressVote`.
+The goal is to minimize vote size while retaining a 1‑to‑1, loss‑free mapping to the canonical msgpack representation of `agreement.UnauthenticatedVote`.
+The canonical msgpack representation we rely on is provided by agreement/msgp_gen.go, generated by our [custom msgpack code generator](https://github.com/algorand/msgp)
+which ensures fields are generated in lexicographic order, omit empty key-value pairs, and use specific formats for certain types as defined in
+[our specification](https://github.com/algorandfoundation/specs/blob/c0331123148971e4705f25b9c937cb23e5ee28d1/dev/crypto.md#L22-L40).
+
+---
+
+## 1. High‑level structure
+
+```
++---------+-----------------+---------------------+--------------------------+
+| Header  | VrfProof ("pf") | rawVote ("r")       | OneTimeSignature ("sig") |
+| 2 bytes | 80 bytes        | variable length     | 256 bytes                |
++---------+-----------------+---------------------+--------------------------+
+```
+
+All fields appear exactly once, and in the fixed order above. The presence of optional sub‑fields inside `rawVote` are indicated by a 1‑byte bitmask in the header.
+No field names appear, only values.
+
+---
+
+## 2. Header (2 bytes)
+
+| Offset | Description                                                    |
+| ------ | -------------------------------------------------------------- |
+| `0`    | Presence flags for optional values (LSB first, see table).     |
+| `1`    | Reserved, currently zero.                                      |
+
+### 2.1 Bit‑mask layout (byte 0)
+
+| Bit | Flag        | Field enabled                    | Encoded size |
+| --- | ----------- | -------------------------------- | ------------ |
+| 0   | `bitPer`    | `r.per` (varuint)                | 1 - 9 bytes  |
+| 1   | `bitDig`    | `r.prop.dig` (digest)            | 32 bytes     |
+| 2   | `bitEncDig` | `r.prop.encdig` (digest)         | 32 bytes     |
+| 3   | `bitOper`   | `r.prop.oper` (varuint)          | 1 - 9 bytes  |
+| 4   | `bitOprop`  | `r.prop.oprop` (address)         | 32 bytes     |
+| 5   | `bitStep`   | `r.step` (varuint)               | 1 - 9 bytes  |
+
+Binary fields are represented by their 32-, 64-, and 80-byte values without markers.
+Integers use msgpack's variable-length unsigned integer encoding:
+- `fixint` (≤ 127), 1 byte in length (values 0x00-0x7f)
+- `uint8` 2 bytes in length (marker byte 0xcc + 1-byte value)
+- `uint16` 3 bytes in length (marker byte 0xcd + 2-byte value)
+- `uint32` 5 bytes in length (marker byte 0xce + 4-byte value)
+- `uint64` 9 bytes in length (marker byte 0xcf + 8-byte value)
+
+---
+
+## 3. Field serialization order
+
+After the 2-byte header, the encoder emits values in the following order:
+
+| Field          | Type                           | Encoded size | Presence flag |
+| -------------- | ------------------------------ | ------------ | ------------- |
+| `pf`           | VRF credential                 | 80 bytes     | Required      |
+| `r.per`        | Period                         | varuint      | `bitPer`      |
+| `r.prop.dig`   | Proposal digest                | 32 bytes     | `bitDig`      |
+| `r.prop.encdig`| Digest of encoded proposal     | 32 bytes     | `bitEncDig`   |
+| `r.prop.oper`  | Proposal's original period     | varuint      | `bitOper`     |
+| `r.prop.oprop` | Proposal's original proposer   | 32 bytes     | `bitOprop`    |
+| `r.rnd`        | Round number                   | varuint      | Required      |
+| `r.snd`        | Voter's (sender) address       | 32 bytes     | Required      |
+| `r.step`       | Step                           | varuint      | `bitStep`     |
+| `sig.p`        | Ed25519 public key             | 32 bytes     | Required      |
+| `sig.p1s`      | Signature over offset ID       | 64 bytes     | Required      |
+| `sig.p2`       | Second-tier Ed25519 public key | 32 bytes     | Required      |
+| `sig.p2s`      | Signature over batch ID        | 64 bytes     | Required      |
+| `sig.s`        | Signature over vote using `p`  | 64 bytes     | Required      |