From 92a824fbb3f6faa47bbb418441c9e43cfb62d294 Mon Sep 17 00:00:00 2001
From: Fabianexe <Fabianexe@gmail.com>
Date: Wed, 4 Nov 2020 02:09:01 +0100
Subject: [PATCH] Fix a bug that obscures the byte queue when it is full and is
 extended (#251)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* Fix a bug that obscures the byte queue when it is full and need to re allocate

* Fix a bug that add artificial entry that is to short to have a valid header

* Fix size check for larger minimum header size

Co-authored-by: Fabian Gärtner <fabian.gaertner@check24.de>
---
 queue/bytes_queue.go      | 13 ++++++++-----
 queue/bytes_queue_test.go | 26 ++++++++++++++++++++++++--
 2 files changed, 32 insertions(+), 7 deletions(-)

diff --git a/queue/bytes_queue.go b/queue/bytes_queue.go
index 184c1e8b..b8e04357 100644
--- a/queue/bytes_queue.go
+++ b/queue/bytes_queue.go
@@ -8,7 +8,7 @@ import (
 
 const (
 	// Number of bytes to encode 0 in uvarint format
-	minimumHeaderSize = 1
+	minimumHeaderSize = 17 // 1 byte blobsize + timestampSizeInBytes + hashSizeInBytes
 	// Bytes before left margin are not used. Zero index means element does not exist in queue, useful while reading slice from index
 	leftMarginIndex = 1
 )
@@ -118,10 +118,13 @@ func (q *BytesQueue) allocateAdditionalMemory(minimum int) {
 	if leftMarginIndex != q.rightMargin {
 		copy(q.array, oldArray[:q.rightMargin])
 
-		if q.tail < q.head {
-			headerEntrySize := getUvarintSize(uint32(q.head - q.tail))
-			emptyBlobLen := q.head - q.tail - headerEntrySize
-			q.push(make([]byte, emptyBlobLen), emptyBlobLen)
+		if q.tail <= q.head {
+			if q.tail != q.head {
+				headerEntrySize := getUvarintSize(uint32(q.head - q.tail))
+				emptyBlobLen := q.head - q.tail - headerEntrySize
+				q.push(make([]byte, emptyBlobLen), emptyBlobLen)
+			}
+
 			q.head = leftMarginIndex
 			q.tail = q.rightMargin
 		}
diff --git a/queue/bytes_queue_test.go b/queue/bytes_queue_test.go
index 545bf7da..6518c28e 100644
--- a/queue/bytes_queue_test.go
+++ b/queue/bytes_queue_test.go
@@ -389,8 +389,6 @@ func TestPushEntryAfterAllocateAdditionMemory(t *testing.T) {
 	queue.Push([]byte("aaa"))
 	queue.Push([]byte("bb"))
 	queue.Pop()
-	queue.Push([]byte("c"))
-	queue.Push([]byte("d"))
 
 	// allocate more memory
 	assertEqual(t, 9, queue.Capacity())
@@ -402,6 +400,30 @@ func TestPushEntryAfterAllocateAdditionMemory(t *testing.T) {
 	noError(t, err)
 }
 
+func TestPushEntryAfterAllocateAdditionMemoryInFull(t *testing.T) {
+	t.Parallel()
+
+	// given
+	queue := NewBytesQueue(9, 40, true)
+
+	// when
+	queue.Push([]byte("aaa"))
+	queue.Push([]byte("bb"))
+	_, err := queue.Pop()
+	noError(t, err)
+
+	queue.Push([]byte("c"))
+	queue.Push([]byte("d"))
+	queue.Push([]byte("e"))
+	_, err = queue.Pop()
+	noError(t, err)
+	_, err = queue.Pop()
+	noError(t, err)
+	queue.Push([]byte("fff"))
+	_, err = queue.Pop()
+	noError(t, err)
+}
+
 func pop(queue *BytesQueue) []byte {
 	entry, err := queue.Pop()
 	if err != nil {