fix: Fix timer logic in batch writers (#1056)

I believe that the batch timeout should be a guarantee of a message being delayed by no longer than X amount of time. This means that we shouldn't reset the ticker every time a message is received, because then a message could be delayed for a very long time if a slow trickle of messages is coming in. This change fixes this issue and guarantees a message to be flushed within at most X duration.

For more context, see this PR and message #1055 (comment)

Author: hermanschaaf

writers/batchwriter/batchwriter.go

@@ -122,6 +122,7 @@ func (w *BatchWriter) Close(context.Context) error {
 func (w *BatchWriter) worker(ctx context.Context, tableName string, ch <-chan *message.WriteInsert, flush <-chan chan bool) {
 	sizeBytes := int64(0)
 	resources := make([]*message.WriteInsert, 0, w.batchSize)
+	tick := timer(w.batchTimeout)
 	for {
 		select {
 		case r, ok := <-ch:
@@ -139,11 +140,12 @@ func (w *BatchWriter) worker(ctx context.Context, tableName string, ch <-chan *m
 
 			resources = append(resources, r)
 			sizeBytes += util.TotalRecordSize(r.Record)
-		case <-timer(w.batchTimeout):
+		case <-tick:
 			if len(resources) > 0 {
 				w.flushTable(ctx, tableName, resources)
 				resources, sizeBytes = resources[:0], 0
 			}
+			tick = timer(w.batchTimeout)
 		case done := <-flush:
 			if len(resources) > 0 {
 				w.flushTable(ctx, tableName, resources)

writers/streamingbatchwriter/streamingbatchwriter.go

@@ -345,6 +345,7 @@ func (s *streamingWorkerManager[T]) run(ctx context.Context, wg *sync.WaitGroup,
 	}
 	defer closeFlush()
 
+	tick := s.timerFn(s.batchTimeout)
 	for {
 		select {
 		case r, ok := <-s.ch:
@@ -365,10 +366,11 @@ func (s *streamingWorkerManager[T]) run(ctx context.Context, wg *sync.WaitGroup,
 			clientCh <- r
 			sizeRows++
 			sizeBytes += recSize
-		case <-s.timerFn(s.batchTimeout):
+		case <-tick:
 			if sizeRows > 0 {
 				closeFlush()
 			}
+			tick = s.timerFn(s.batchTimeout)
 		case done := <-s.flush:
 			if sizeRows > 0 {
 				closeFlush()