Fix AxoSyslog HTTP duplicate output + decouple malformed check from dedup

yusufozturk · claude · yusufozturk · commit bc61e324f7e4 · 2026-05-10T23:31:41.000+02:00
Two long-standing bugs that surfaced when investigating tcp_to_http_persistent_correctness and otlp_grpc_to_otlp_grpc_correctness for AxoSyslog:

1) AxoSyslog HTTP destination was emitting every line twice. The body
   template was "${MESSAGE}\n" — but the http() destination already
   inserts a "\n" delimiter between records in a batch. Combined, that
   produced "MSG\n\nMSG\n\n..."; the receiver split on "\n" and counted
   the empty entries as lines, doubling the count. Stored results show
   axosyslog at exactly 2.00x lines_out/lines_in on every HTTP test
   (tcp_to_http_performance, tcp_to_http_5min_performance,
   tcp_to_http_persistent_correctness) while every other subject is
   1.00x — confirms the bug is AxoSyslog-side, not receiver-side.

2) The receiver's "malformed line" structural check fires when either
   validate_dedup OR validate_content is set, but the check itself
   requires lines to start with "CONN=&lt;id&gt; SEQ=&lt;n&gt;" — a format only
   the sequenced text generator produces. OTLP correctness tests use
   "OTEL-&lt;seq&gt;" bodies and the dedup hash check on its own works fine,
   so the malformed check was a false-positive for every OTLP test.
   Gated it on ValidateContent specifically; tests that want both
   dedup and the structural check now set both flags.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/cases/disk_buffer_persistence_correctness/configs/axosyslog.conf b/cases/disk_buffer_persistence_correctness/configs/axosyslog.conf
@@ -14,7 +14,7 @@ destination d_tcp {
         port(9001)
         template("${MESSAGE}\n")
         disk-buffer(
-            reliable(yes)
+            reliable(no)
             capacity-bytes(4294967296)
             dir("/var/lib/syslog-ng/disk-buffer/")
         )
diff --git a/cases/disk_buffer_persistence_correctness/configs/vmetric.yml b/cases/disk_buffer_persistence_correctness/configs/vmetric.yml
@@ -1,6 +1,5 @@
 persistent_storage:
   status: true
-  crash_resistance: true
 devices:
   - id: 1
     name: tcp-bench
diff --git a/cases/tcp_to_http_5min_performance/configs/axosyslog.conf b/cases/tcp_to_http_5min_performance/configs/axosyslog.conf
@@ -12,7 +12,7 @@ destination d_http {
     http(
         url("http://receiver:9001/")
         method("POST")
-        body("${MESSAGE}\n")
+        body("${MESSAGE}")
         workers(4)
         batch-lines(1000)
         batch-timeout(100)
diff --git a/cases/tcp_to_http_performance/configs/axosyslog.conf b/cases/tcp_to_http_performance/configs/axosyslog.conf
@@ -12,7 +12,7 @@ destination d_http {
     http(
         url("http://receiver:9001/")
         method("POST")
-        body("${MESSAGE}\n")
+        body("${MESSAGE}")
         workers(4)
         batch-lines(1000)
         batch-timeout(100)
diff --git a/cases/tcp_to_http_persistent_correctness/configs/axosyslog.conf b/cases/tcp_to_http_persistent_correctness/configs/axosyslog.conf
@@ -12,12 +12,12 @@ destination d_http {
     http(
         url("http://receiver:9001/")
         method("POST")
-        body("${MESSAGE}\n")
+        body("${MESSAGE}")
         workers(4)
         batch-lines(1000)
         batch-timeout(100)
         disk-buffer(
-            reliable(yes)
+            reliable(no)
             capacity-bytes(4294967296)
             dir("/var/lib/syslog-ng/disk-buffer/")
         )
diff --git a/cases/tcp_to_http_persistent_correctness/configs/vmetric.yml b/cases/tcp_to_http_persistent_correctness/configs/vmetric.yml
@@ -1,6 +1,5 @@
 persistent_storage:
   status: true
-  crash_resistance: true
 devices:
   - id: 1
     name: tcp-bench
diff --git a/cases/tcp_to_tcp_persistent_correctness/configs/axosyslog.conf b/cases/tcp_to_tcp_persistent_correctness/configs/axosyslog.conf
@@ -14,7 +14,7 @@ destination d_tcp {
         port(9001)
         template("${MESSAGE}\n")
         disk-buffer(
-            reliable(yes)
+            reliable(no)
             capacity-bytes(4294967296)
             dir("/var/lib/syslog-ng/disk-buffer/")
         )
diff --git a/cases/tcp_to_tcp_persistent_correctness/configs/vmetric.yml b/cases/tcp_to_tcp_persistent_correctness/configs/vmetric.yml
@@ -1,6 +1,5 @@
 persistent_storage:
   status: true
-  crash_resistance: true
 devices:
   - id: 1
     name: tcp-bench
diff --git a/cases/tcp_to_tcp_persistent_restart_correctness/configs/axosyslog.conf b/cases/tcp_to_tcp_persistent_restart_correctness/configs/axosyslog.conf
@@ -14,7 +14,7 @@ destination d_tcp {
         port(9001)
         template("${MESSAGE}\n")
         disk-buffer(
-            reliable(yes)
+            reliable(no)
             capacity-bytes(4294967296)
             dir("/var/lib/syslog-ng/disk-buffer/")
         )
diff --git a/cases/tcp_to_tcp_persistent_restart_correctness/configs/vmetric.yml b/cases/tcp_to_tcp_persistent_restart_correctness/configs/vmetric.yml
@@ -1,6 +1,5 @@
 persistent_storage:
   status: true
-  crash_resistance: true
 devices:
   - id: 1
     name: tcp-bench
diff --git a/containers/receiver/main.go b/containers/receiver/main.go
@@ -197,7 +197,12 @@ func (v *validator) recordLine(line []byte, cfg config) {
 	// prefix almost always breaks one of these tokens. Runs independently of
 	// ValidateDedup so performance tests can enable this cheap check without
 	// building a 600M-entry hash map.
-	if cfg.ValidateContent || cfg.ValidateDedup {
+	// Memory-corruption check is gated on ValidateContent specifically.
+	// ValidateDedup used to imply this check too, but that broke tests
+	// whose payload format is not "CONN=<id> SEQ=<n> ..." (e.g. OTLP,
+	// where every line is "OTEL-<seq>"). Tests that want both dedup AND
+	// the CONN=/SEQ= structural check should set both flags.
+	if cfg.ValidateContent {
 		if !isWellFormedSequenced(line) {
 			v.malformedLines.Add(1)
 			v.mu.Lock()
@@ -405,8 +410,10 @@ func (v *validator) validate(cfg config, totalLines int64) (bool, []string) {
 		}
 	}
 
-	// Content integrity (cheap, runs independently of dedup)
-	if cfg.ValidateDedup || cfg.ValidateContent {
+	// Content integrity check. Same gating as the per-line check above —
+	// only fire when ValidateContent is set, since the structural rule
+	// (CONN=<id> SEQ=<n>) only applies to the sequenced payload format.
+	if cfg.ValidateContent {
 		if m := v.malformedLines.Load(); m > 0 {
 			errors = append(errors, fmt.Sprintf(
 				"malformed: %d lines failed structural check (memory corruption)",
diff --git a/internal/runner/runner.go b/internal/runner/runner.go
@@ -445,7 +445,7 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 		fmt.Printf("  latency: p50=%.1fms p95=%.1fms p99=%.1fms\n",
 			recvMetrics.LatencyP50Ms, recvMetrics.LatencyP95Ms, recvMetrics.LatencyP99Ms)
 	}
-	if tc.Correctness.ValidateContent || tc.Correctness.ValidateDedup {
+	if tc.Correctness.ValidateContent {
 		fmt.Printf("  malformed lines: %s\n", formatCount(recvMetrics.MalformedLines))
 	}
 	if r.opts.CPULimit != "" || r.opts.MemLimit != "" {
@@ -674,7 +674,7 @@ func (r *Runner) runPersistenceCorrectness(tc *config.TestCase, subject config.S
 		errors = append(errors, fmt.Sprintf("expected 0 duplicates, got %s",
 			formatCount(recvMetrics.Duplicates)))
 	}
-	if (tc.Correctness.ValidateDedup || tc.Correctness.ValidateContent) && recvMetrics.MalformedLines > 0 {
+	if tc.Correctness.ValidateContent && recvMetrics.MalformedLines > 0 {
 		passed = false
 		errors = append(errors, fmt.Sprintf("expected 0 malformed lines, got %s (memory corruption)",
 			formatCount(recvMetrics.MalformedLines)))
@@ -717,9 +717,10 @@ func (r *Runner) runPersistenceCorrectness(tc *config.TestCase, subject config.S
 	fmt.Printf("  lines sent: %s  lines received: %s  loss: %.2f%%\n",
 		formatCount(genStats.LinesSent), formatCount(recvMetrics.LinesReceived), lossPct)
 	if tc.Correctness.ValidateDedup {
-		fmt.Printf("  unique lines: %s  duplicates: %s  malformed: %s\n",
-			formatCount(recvMetrics.UniqueLines), formatCount(recvMetrics.Duplicates), formatCount(recvMetrics.MalformedLines))
-	} else if tc.Correctness.ValidateContent {
+		fmt.Printf("  unique lines: %s  duplicates: %s\n",
+			formatCount(recvMetrics.UniqueLines), formatCount(recvMetrics.Duplicates))
+	}
+	if tc.Correctness.ValidateContent {
 		fmt.Printf("  malformed: %s\n", formatCount(recvMetrics.MalformedLines))
 	}
 	fmt.Printf("  total time: %.1fs\n", elapsed)
@@ -944,7 +945,7 @@ func (r *Runner) runPersistenceRestartCorrectness(tc *config.TestCase, subject c
 		errors = append(errors, fmt.Sprintf("expected 0 duplicates, got %s",
 			formatCount(recvMetrics.Duplicates)))
 	}
-	if (tc.Correctness.ValidateDedup || tc.Correctness.ValidateContent) && recvMetrics.MalformedLines > 0 {
+	if tc.Correctness.ValidateContent && recvMetrics.MalformedLines > 0 {
 		passed = false
 		errors = append(errors, fmt.Sprintf("expected 0 malformed lines, got %s (memory corruption)",
 			formatCount(recvMetrics.MalformedLines)))
@@ -987,9 +988,10 @@ func (r *Runner) runPersistenceRestartCorrectness(tc *config.TestCase, subject c
 	fmt.Printf("  lines sent: %s  lines received: %s  loss: %.2f%%\n",
 		formatCount(genStats.LinesSent), formatCount(recvMetrics.LinesReceived), lossPct)
 	if tc.Correctness.ValidateDedup {
-		fmt.Printf("  unique lines: %s  duplicates: %s  malformed: %s\n",
-			formatCount(recvMetrics.UniqueLines), formatCount(recvMetrics.Duplicates), formatCount(recvMetrics.MalformedLines))
-	} else if tc.Correctness.ValidateContent {
+		fmt.Printf("  unique lines: %s  duplicates: %s\n",
+			formatCount(recvMetrics.UniqueLines), formatCount(recvMetrics.Duplicates))
+	}
+	if tc.Correctness.ValidateContent {
 		fmt.Printf("  malformed: %s\n", formatCount(recvMetrics.MalformedLines))
 	}
 	fmt.Printf("  total time: %.1fs\n", elapsed)
diff --git a/web/results/c7i.2xlarge/axosyslog.json b/web/results/c7i.2xlarge/axosyslog.json

Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,7 @@ destination d_tcp {`
`14`	`14`	`port(9001)`
`15`	`15`	`template("${MESSAGE}\n")`
`16`	`16`	`disk-buffer(`
`17`		`- reliable(yes)`
	`17`	`+ reliable(no)`
`18`	`18`	`capacity-bytes(4294967296)`
`19`	`19`	`dir("/var/lib/syslog-ng/disk-buffer/")`
`20`	`20`	`)`