network: stateless vote compression

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 @@
 	"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 @@
 	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 @@
 	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 (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,32 @@
 			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{}
+	if 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,65 @@
+# 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`.
+
+---
+
+## 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.
+
+---
+
+## 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                    | Size needed |
+| --- | ----------- | -------------------------------- | ----------- |
+| 0   | `bitPer`    | `r.per` (varuint)                | 1 – 9       |
+| 1   | `bitDig`    | `r.prop.dig` (32-byte digest)    | 32          |
+| 2   | `bitEncDig` | `r.prop.encdig` (32-byte digest) | 32          |
+| 3   | `bitOper`   | `r.prop.oper` (varuint)          | 1 – 9       |
+| 4   | `bitOprop`  | `r.prop.oprop` (32-byte address) | 32          |
+| 5   | `bitStep`   | `r.step` (varuint)               | 1 - 9       |
+
+*Variable‑length integers* use msgpack varuint encoding:
+- `fixint` (≤ 127), 1 byte in length
+- `uint8` 2 bytes in length (1 for marker, 1 for value)
+- `uint16` 3 bytes in length (1 for marker, 2 for value)
+- `uint32` 5 bytes in length (1 for marker, 4 for value)
+- `uint64` 9 bytes in length (1 for marker, 8 for value)
+
+---
+
+## 3. Field serialization order
+
+After the 2-byte header, the encoder emits values in the following order:
+
+1. `pf` VRF credential (80 bytes), always present.
+1. `r.per` period (varuint), if `bitPer` is set.
+1. `r.prop.dig` proposal's digest (32 bytes), if `bitDig` is set.
+1. `r.prop.encdig` encoded proposal's digest (32 bytes), if `bitEncDig` is set.
+1. `r.prop.oper` proposal's original period (32 bytes), if `bitOper` is set.
+1. `r.prop.oprop` proposal's original proposer (32 bytes), if `bitOprop` is set.
+1. `r.rnd` round number (varuint), always present.
+1. `r.snd` sender address (32 bytes) always present.
+1. `r.step` step (varuint), if `bitStep` is set.
+1. `sig.p` public key (32 bytes), always present.
+1. `sig.p1s` signature of offset ID (64 bytes), always present.
+1. `sig.p2` second-tier public key (32 bytes), always present.
+1. `sig.p2s` signature of batch ID (64 bytes), always present.
+1. `sig.s` signature of message under key p (64 bytes), always present.