[-] make lastError sends in flush() non-blocking, closes #1212 (#1265)

pashagolub · web-flow · commit d22ca40964e5 · 2026-03-02T10:39:34.000+01:00
diff --git a/internal/sinks/postgres.go b/internal/sinks/postgres.go
@@ -422,7 +422,10 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {
 	}
 	pgw.forceRecreatePartitions = false
 	if err != nil {
-		pgw.lastError <- err
+		select {
+		case pgw.lastError <- err:
+		default:
+		}
 	}
 
 	var rowsBatched, n int64
@@ -446,7 +449,10 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {
 		logger.WithField("rows", rowsBatched).WithField("elapsed", diff).Info("measurements written")
 		return
 	}
-	pgw.lastError <- err
+	select {
+	case pgw.lastError <- err:
+	default:
+	}
 }
 
 func (pgw *PostgresWriter) EnsureMetricTimescale(pgPartBounds map[string]ExistingPartitionInfo) (err error) {

Original file line number	Diff line number	Diff line change
`@@ -422,7 +422,10 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {`
`422`	`422`	`}`
`423`	`423`	`pgw.forceRecreatePartitions = false`
`424`	`424`	`if err != nil {`
`425`		`- pgw.lastError <- err`
	`425`	`+ select {`
	`426`	`+ case pgw.lastError <- err:`
	`427`	`+ default:`
	`428`	`+ }`
`426`	`429`	`}`
`427`	`430`
`428`	`431`	`var rowsBatched, n int64`
`@@ -446,7 +449,10 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {`
`446`	`449`	`logger.WithField("rows", rowsBatched).WithField("elapsed", diff).Info("measurements written")`
`447`	`450`	`return`
`448`	`451`	`}`
`449`		`- pgw.lastError <- err`
	`452`	`+ select {`
	`453`	`+ case pgw.lastError <- err:`
	`454`	`+ default:`
	`455`	`+ }`
`450`	`456`	`}`
`451`	`457`
`452`	`458`	`func (pgw *PostgresWriter) EnsureMetricTimescale(pgPartBounds map[string]ExistingPartitionInfo) (err error) {`