Add MMD verification in CT Hammer

Author: roger2hk

deployment/modules/gcp/cloudbuild/main.tf

@@ -173,6 +173,7 @@ resource "google_cloudbuild_trigger" "build_trigger" {
           --logtostderr \
           --num_writers=256 \
           --max_write_ops=256 \
+          --num_mmd_verifiers=256 \
           --leaf_write_goal=10000
       EOT
       wait_for = ["bearer_token"]

internal/hammer/README.md

@@ -17,14 +17,15 @@ Example usage to test a deployment of `cmd/gcp`:
 
 ```shell
 go run ./internal/hammer \
-  --log_public_key=test-static-ct+59739ea1+BTBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABGbaLj7T8pSEfEYL6nbF8U1xLjoy+dBkL5pINuSaTZ6DTW2WQ1bdZ4lO8ZuAcGLtSRESI01di5ZskWwgRwphuiY= \
+  --log_public_key=MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEZtouPtPylIR8RgvqdsXxTXEuOjL50GQvmkg25JpNnoNNbZZDVt1niU7xm4BwYu1JERIjTV2LlmyRbCBHCmG6Jg== \
   --log_url=https://storage.googleapis.com/transparency-dev-playground-test-static-ct-bucket \
   --write_log_url=http://localhost:6962/test-static-ct
   --max_read_ops=1024 \
   --num_readers_random=128 \
   --num_readers_full=128 \
   --num_writers=256 \
   --max_write_ops=42 \
+  --num_mmd_verifiers=256 \
   --bearer_token=$(gcloud auth print-access-token)
 ```
 
@@ -34,12 +35,13 @@ If the timeout of 1 minute is reached first, then it exits with an exit code of
 
 ```shell
 go run ./internal/hammer \
-  --log_public_key=test-static-ct+59739ea1+BTBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABGbaLj7T8pSEfEYL6nbF8U1xLjoy+dBkL5pINuSaTZ6DTW2WQ1bdZ4lO8ZuAcGLtSRESI01di5ZskWwgRwphuiY= \
+  --log_public_key=MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEZtouPtPylIR8RgvqdsXxTXEuOjL50GQvmkg25JpNnoNNbZZDVt1niU7xm4BwYu1JERIjTV2LlmyRbCBHCmG6Jg== \
   --log_url=https://storage.googleapis.com/transparency-dev-playground-test-static-ct-bucket \
   --write_log_url=http://localhost:6962/test-static-ct
   --max_read_ops=0 \
   --num_writers=512 \
   --max_write_ops=512 \
+  --num_mmd_verifiers=512 \
   --max_runtime=1m \
   --leaf_write_goal=2500 \
   --bearer_token=$(gcloud auth print-access-token) \

internal/hammer/hammer.go

@@ -65,6 +65,7 @@ var (
 
 	maxWriteOpsPerSecond = flag.Int("max_write_ops", 0, "The maximum number of write operations per second")
 	numWriters           = flag.Int("num_writers", 0, "The number of independent write tasks to run")
+	numMMDVerifiers      = flag.Int("num_mmd_verifiers", 0, "The number of MMD verifiers performing inclusion proof for the added leaves")
 
 	dupChance = flag.Float64("dup_chance", 0.1, "The probability of a generated leaf being a duplicate of a previous value")
 
@@ -160,6 +161,7 @@ func main() {
 		NumReadersRandom:     *numReadersRandom,
 		NumReadersFull:       *numReadersFull,
 		NumWriters:           *numWriters,
+		NumMMDVerifiers:      *numMMDVerifiers,
 	}
 	hammer := loadtest.NewHammer(&tracker, f.ReadEntryBundle, w, gen, ha.SeqLeafChan, ha.ErrChan, opts)
 

internal/hammer/loadtest/client.go

@@ -155,42 +155,42 @@ type httpLeafWriter struct {
 	bearerToken string
 }
 
-func (w httpLeafWriter) Write(ctx context.Context, newLeaf []byte) (uint64, error) {
+func (w httpLeafWriter) Write(ctx context.Context, newLeaf []byte) (uint64, uint64, error) {
 	req, err := http.NewRequest(http.MethodPost, w.u.String(), bytes.NewReader(newLeaf))
 	if err != nil {
-		return 0, fmt.Errorf("failed to create request: %v", err)
+		return 0, 0, fmt.Errorf("failed to create request: %v", err)
 	}
 	if w.bearerToken != "" {
 		req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", w.bearerToken))
 	}
 	resp, err := w.hc.Do(req.WithContext(ctx))
 	if err != nil {
-		return 0, fmt.Errorf("failed to write leaf: %v", err)
+		return 0, 0, fmt.Errorf("failed to write leaf: %v", err)
 	}
 	body, err := io.ReadAll(resp.Body)
 	_ = resp.Body.Close()
 	if err != nil {
-		return 0, fmt.Errorf("failed to read body: %v", err)
+		return 0, 0, fmt.Errorf("failed to read body: %v", err)
 	}
 	switch resp.StatusCode {
 	case http.StatusOK:
 		if resp.Request.Method != http.MethodPost {
-			return 0, fmt.Errorf("write leaf was redirected to %s", resp.Request.URL)
+			return 0, 0, fmt.Errorf("write leaf was redirected to %s", resp.Request.URL)
 		}
 		// Continue below
 	case http.StatusServiceUnavailable, http.StatusBadGateway, http.StatusGatewayTimeout:
 		// These status codes may indicate a delay before retrying, so handle that here:
 		time.Sleep(retryDelay(resp.Header.Get("RetryAfter"), time.Second))
 
-		return 0, fmt.Errorf("log not available. Status code: %d. Body: %q %w", resp.StatusCode, body, ErrRetry)
+		return 0, 0, fmt.Errorf("log not available. Status code: %d. Body: %q %w", resp.StatusCode, body, ErrRetry)
 	default:
-		return 0, fmt.Errorf("write leaf was not OK. Status code: %d. Body: %q", resp.StatusCode, body)
+		return 0, 0, fmt.Errorf("write leaf was not OK. Status code: %d. Body: %q", resp.StatusCode, body)
 	}
-	index, err := parseAddChainResponse(body)
+	index, timestamp, err := parseAddChainResponse(body)
 	if err != nil {
-		return 0, fmt.Errorf("write leaf failed to parse response: %v", body)
+		return 0, 0, fmt.Errorf("write leaf failed to parse response: %v", body)
 	}
-	return index, nil
+	return index, timestamp, nil
 }
 
 func retryDelay(retryAfter string, defaultDur time.Duration) time.Duration {
@@ -216,7 +216,7 @@ type roundRobinLeafWriter struct {
 	ws  []httpLeafWriter
 }
 
-func (rr *roundRobinLeafWriter) Write(ctx context.Context, newLeaf []byte) (uint64, error) {
+func (rr *roundRobinLeafWriter) Write(ctx context.Context, newLeaf []byte) (uint64, uint64, error) {
 	w := rr.next()
 	return w(ctx, newLeaf)
 }
@@ -232,39 +232,39 @@ func (rr *roundRobinLeafWriter) next() LeafWriter {
 }
 
 // parseAddChainResponse parses the add-chain response and returns the leaf
-// index from the extensions.
+// index from the extensions and timestamp from the response.
 // Code is inspired by https://github.com/FiloSottile/sunlight/blob/main/tile.go.
-func parseAddChainResponse(body []byte) (uint64, error) {
+func parseAddChainResponse(body []byte) (uint64, uint64, error) {
 	var resp types.AddChainResponse
 	if err := json.Unmarshal(body, &resp); err != nil {
-		return 0, fmt.Errorf("can't parse add-chain response: %v", err)
+		return 0, 0, fmt.Errorf("can't parse add-chain response: %v", err)
 	}
 
 	extensionBytes, err := base64.StdEncoding.DecodeString(resp.Extensions)
 	if err != nil {
-		return 0, fmt.Errorf("can't decode extensions: %v", err)
+		return 0, 0, fmt.Errorf("can't decode extensions: %v", err)
 	}
 	extensions := cryptobyte.String(extensionBytes)
 	var extensionType uint8
 	var extensionData cryptobyte.String
 	var leafIdx int64
 	if !extensions.ReadUint8(&extensionType) {
-		return 0, fmt.Errorf("can't read extension type")
+		return 0, 0, fmt.Errorf("can't read extension type")
 	}
 	if extensionType != 0 {
-		return 0, fmt.Errorf("wrong extension type %d, want 0", extensionType)
+		return 0, 0, fmt.Errorf("wrong extension type %d, want 0", extensionType)
 	}
 	if !extensions.ReadUint16LengthPrefixed(&extensionData) {
-		return 0, fmt.Errorf("can't read extension data")
+		return 0, 0, fmt.Errorf("can't read extension data")
 	}
 	if !readUint40(&extensionData, &leafIdx) {
-		return 0, fmt.Errorf("can't read leaf index from extension")
+		return 0, 0, fmt.Errorf("can't read leaf index from extension")
 	}
 	if !extensionData.Empty() ||
 		!extensions.Empty() {
-		return 0, fmt.Errorf("invalid data tile extensions: %v", resp.Extensions)
+		return 0, 0, fmt.Errorf("invalid data tile extensions: %v", resp.Extensions)
 	}
-	return uint64(leafIdx), nil
+	return uint64(leafIdx), resp.Timestamp, nil
 }
 
 // readUint40 decodes a big-endian, 40-bit value into out and advances over it.

internal/hammer/loadtest/hammer.go

@@ -31,27 +31,34 @@ type HammerOpts struct {
 	NumReadersRandom int
 	NumReadersFull   int
 	NumWriters       int
+	NumMMDVerifiers  int
 }
 
 func NewHammer(tracker *client.LogStateTracker, f client.EntryBundleFetcherFunc, w LeafWriter, gen func() []byte, seqLeafChan chan<- LeafTime, errChan chan<- error, opts HammerOpts) *Hammer {
 	readThrottle := NewThrottle(opts.MaxReadOpsPerSecond)
 	writeThrottle := NewThrottle(opts.MaxWriteOpsPerSecond)
 
+	leafMMDChan := make(chan LeafMMD, opts.NumWriters)
+
 	randomReaders := NewWorkerPool(func() Worker {
 		return NewLeafReader(tracker, f, RandomNextLeaf(), readThrottle.TokenChan, errChan)
 	})
 	fullReaders := NewWorkerPool(func() Worker {
 		return NewLeafReader(tracker, f, MonotonicallyIncreasingNextLeaf(), readThrottle.TokenChan, errChan)
 	})
 	writers := NewWorkerPool(func() Worker {
-		return NewLogWriter(w, gen, writeThrottle.TokenChan, errChan, seqLeafChan)
+		return NewLogWriter(w, gen, writeThrottle.TokenChan, errChan, seqLeafChan, leafMMDChan)
+	})
+	mmdVerifiers := NewWorkerPool(func() Worker {
+		return NewMMDVerifier(tracker, errChan, leafMMDChan)
 	})
 
 	return &Hammer{
 		opts:          opts,
 		randomReaders: randomReaders,
 		fullReaders:   fullReaders,
 		writers:       writers,
+		mmdVerifiers:  mmdVerifiers,
 		readThrottle:  readThrottle,
 		writeThrottle: writeThrottle,
 		tracker:       tracker,
@@ -66,6 +73,7 @@ type Hammer struct {
 	randomReaders WorkerPool
 	fullReaders   WorkerPool
 	writers       WorkerPool
+	mmdVerifiers  WorkerPool
 	readThrottle  *Throttle
 	writeThrottle *Throttle
 	tracker       *client.LogStateTracker
@@ -82,6 +90,9 @@ func (h *Hammer) Run(ctx context.Context) {
 	for i := 0; i < h.opts.NumWriters; i++ {
 		h.writers.Grow(ctx)
 	}
+	for i := 0; i < h.opts.NumMMDVerifiers; i++ {
+		h.mmdVerifiers.Grow(ctx)
+	}
 
 	go h.readThrottle.Run(ctx)
 	go h.writeThrottle.Run(ctx)

internal/hammer/loadtest/tui.go

@@ -98,11 +98,13 @@ func (c *tuiController) Run(ctx context.Context) {
 			c.hammer.randomReaders.Grow(ctx)
 			c.hammer.fullReaders.Grow(ctx)
 			c.hammer.writers.Grow(ctx)
+			c.hammer.mmdVerifiers.Grow(ctx)
 		case 'W':
 			klog.Info("Decreasing the number of workers")
 			c.hammer.randomReaders.Shrink(ctx)
 			c.hammer.fullReaders.Shrink(ctx)
 			c.hammer.writers.Shrink(ctx)
+			c.hammer.mmdVerifiers.Shrink(ctx)
 		}
 		return event
 	})