From 9b763af5a660fa6ddfe42fce76b7ada57b83a580 Mon Sep 17 00:00:00 2001
From: Aaron Son <aaron@dolthub.com>
Date: Wed, 24 Mar 2021 15:45:58 -0700
Subject: [PATCH] libraries/utils/async/ring_buffer.go: PR feedback. Return
 error on epoch mismatch.

---
 go/libraries/utils/async/ring_buffer.go      |  7 ++++++-
 go/libraries/utils/async/ring_buffer_test.go | 17 +++++++++++++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/go/libraries/utils/async/ring_buffer.go b/go/libraries/utils/async/ring_buffer.go
index 2910ba0d10..4bd0e19644 100644
--- a/go/libraries/utils/async/ring_buffer.go
+++ b/go/libraries/utils/async/ring_buffer.go
@@ -15,6 +15,7 @@
 package async
 
 import (
+	"errors"
 	"io"
 	"os"
 	"sync"
@@ -35,6 +36,10 @@ type RingBuffer struct {
 	items [][]interface{}
 }
 
+// Returned from Push() when the supplied epoch does not match the
+// buffer's current epoch.
+var ErrWrongEpoch = errors.New("ring buffer: wrong epoch")
+
 // NewRingBuffer creates a new RingBuffer instance
 func NewRingBuffer(allocSize int) *RingBuffer {
 	itemBuffer := make([]interface{}, allocSize*2)
@@ -93,7 +98,7 @@ func (rb *RingBuffer) Push(item interface{}, epoch int) error {
 		return os.ErrClosed
 	}
 	if epoch != rb.epoch {
-		return nil
+		return ErrWrongEpoch
 	}
 
 	rb.items[rb.headSlice][rb.headPos] = item
diff --git a/go/libraries/utils/async/ring_buffer_test.go b/go/libraries/utils/async/ring_buffer_test.go
index 62b5729a39..f2c99a290f 100644
--- a/go/libraries/utils/async/ring_buffer_test.go
+++ b/go/libraries/utils/async/ring_buffer_test.go
@@ -161,3 +161,20 @@ func TestNProducersNConsumers(t *testing.T) {
 		})
 	}
 }
+
+func TestRingBufferEpoch(t *testing.T) {
+	rb := NewRingBuffer(1024)
+	epoch := rb.Reset()
+	err := rb.Push(1, epoch)
+	assert.NoError(t, err)
+	err = rb.Push(2, epoch+1)
+	assert.Error(t, err)
+	assert.Equal(t, ErrWrongEpoch, err)
+	v, ok := rb.TryPop()
+	assert.True(t, ok)
+	assert.Equal(t, 1, v)
+	_, ok = rb.TryPop()
+	assert.False(t, ok)
+	newEpoch := rb.Reset()
+	assert.NotEqual(t, epoch, newEpoch)
+}