From da8b644f40406ee3182d6a61861f7a0c0bc530a3 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Thu, 12 Feb 2026 09:55:08 -0600
Subject: [PATCH] batch SNMP results before sending over WebSocket

Accumulate SNMP results and flush in batches of 50 or every 100ms,
whichever comes first. Reduces per-message overhead when thousands
of devices produce results near-simultaneously. Also flushes
remaining results on disconnect/shutdown.
---
 agent.go | 26 ++++++++++++++++++++++++--
 1 file changed, 24 insertions(+), 2 deletions(-)

diff --git a/agent.go b/agent.go
index 3c40e00..b5773ab 100644
--- a/agent.go
+++ b/agent.go
@@ -173,6 +173,8 @@ func runSession(ctx context.Context, baseURL, token string) error {
 	defer heartbeatTicker.Stop()
 	channelHeartbeatTicker := time.NewTicker(25 * time.Second)
 	defer channelHeartbeatTicker.Stop()
+	flushTicker := time.NewTicker(100 * time.Millisecond)
+	defer flushTicker.Stop()
 	startTime := time.Now()
 
 	defer func() {
@@ -183,13 +185,28 @@ func runSession(ctx context.Context, baseURL, token string) error {
 		writerWg.Wait()
 	}()
 
+	var snmpBatch []*pb.SnmpResult
+
+	flushSnmpBatch := func() {
+		if len(snmpBatch) == 0 {
+			return
+		}
+		for _, r := range snmpBatch {
+			sendBinaryResult("result", r)
+		}
+		slog.Info("flushed snmp results", "count", len(snmpBatch))
+		snmpBatch = snmpBatch[:0]
+	}
+
 	for {
 		select {
 		case <-ctx.Done():
 			slog.Info("shutdown signal, closing connection")
+			flushSnmpBatch()
 			return nil
 
 		case err := <-errCh:
+			flushSnmpBatch()
 			return fmt.Errorf("read: %w", err)
 
 		case data := <-msgCh:
@@ -201,8 +218,10 @@ func runSession(ctx context.Context, baseURL, token string) error {
 			handleMessage(msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh)
 
 		case result := <-snmpResultCh:
-			sendBinaryResult("result", result)
-			slog.Info("sent snmp result", "device", result.DeviceId, "oids", len(result.OidValues))
+			snmpBatch = append(snmpBatch, result)
+			if len(snmpBatch) >= 50 {
+				flushSnmpBatch()
+			}
 
 		case result := <-mikrotikResultCh:
 			sendBinaryResult("mikrotik_result", result)
@@ -216,6 +235,9 @@ func runSession(ctx context.Context, baseURL, token string) error {
 			sendBinaryResult("monitoring_check", result)
 			slog.Info("sent monitoring check", "device", result.DeviceId, "status", result.Status)
 
+		case <-flushTicker.C:
+			flushSnmpBatch()
+
 		case <-heartbeatTicker.C:
 			hb := &pb.AgentHeartbeat{
 				Version:       version,