fix(619): append into ReadWriteSegment when it exceeded size should n…

…ot panic (#627)

Fix: #619

---------

Co-authored-by: Matteo Merli <mmerli@apache.org>

server/wal/wal.go

@@ -28,6 +28,7 @@ var (
 	ErrEntryNotFound     = errors.New("oxia: entry not found")
 	ErrReaderClosed      = errors.New("oxia: reader already closed")
 	ErrInvalidNextOffset = errors.New("oxia: invalid next offset in wal")
+	ErrSegmentFull       = errors.New("oxia: current segment is full")
 
 	InvalidTerm   int64 = -1
 	InvalidOffset int64 = -1

server/wal/wal_impl.go

@@ -284,17 +284,23 @@ func (t *wal) AppendAsync(entry *proto.LogEntry) error {
 		}
 	}
 
-	if !t.currentSegment.HasSpace(len(val)) {
+	if err = t.currentSegment.Append(entry.Offset, val); err != nil {
+		if !errors.Is(err, ErrSegmentFull) {
+			t.writeErrors.Inc()
+			return err
+		}
 		if err = t.rolloverSegment(); err != nil {
 			t.writeErrors.Inc()
 			return errors.Wrap(err, "failed to rollover segment")
 		}
-	}
 
-	if err = t.currentSegment.Append(entry.Offset, val); err != nil {
-		t.writeErrors.Inc()
-		return err
+		// After the rollover, try to append again
+		if err = t.currentSegment.Append(entry.Offset, val); err != nil {
+			t.writeErrors.Inc()
+			return err
+		}
 	}
+
 	t.lastAppendedOffset.Store(entry.Offset)
 	t.firstOffset.CompareAndSwap(InvalidOffset, entry.Offset)
 

server/wal/wal_rw_segment.go

@@ -154,6 +154,9 @@ func (ms *readWriteSegment) Append(offset int64, data []byte) error {
 	if len(data) == 0 {
 		return codec.ErrEmptyPayload
 	}
+	if !ms.HasSpace(len(data)) {
+		return ErrSegmentFull
+	}
 	if offset != ms.lastOffset+1 {
 		return ErrInvalidNextOffset
 	}

server/wal/wal_rw_segment_test.go

@@ -16,6 +16,7 @@ package wal
 
 import (
 	"encoding/binary"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -296,3 +297,16 @@ func TestReadWriteSegment_BrokenCommittedData_ErrDataCorrupted(t *testing.T) {
 	_, err = newReadWriteSegment(dir, 0, 1024, 0, commitOffsetProvider)
 	assert.ErrorIs(t, err, codec.ErrDataCorrupted)
 }
+
+func TestSegmentAppendShouldNotPanic(t *testing.T) {
+	basePath := t.TempDir()
+	rw, err := newReadWriteSegment(basePath, 0, 1024, 0, nil)
+	assert.NoError(t, err)
+	for i := int64(0); i < 51; i++ {
+		err := rw.Append(i, fmt.Appendf(nil, "entry-%d", i))
+		assert.NoError(t, err)
+	}
+
+	err = rw.Append(51, fmt.Appendf(nil, "entry-%d", 51))
+	assert.ErrorIs(t, err, ErrSegmentFull)
+}

server/wal/wal_trimmer_test.go

@@ -101,7 +101,10 @@ func TestWalTrimUpToCommitOffset(t *testing.T) {
 				Retention:   2 * time.Millisecond,
 				SegmentSize: 128 * 1024,
 			}
-
+			slog.Info("Starting",
+				slog.String("TestName", t.Name()),
+				slog.String("BaseWalDir", t.TempDir()),
+			)
 			clock := &common.MockedClock{}
 			commitOffsetProvider := &mockedCommitOffsetProvider{}
 			commitOffsetProvider.commitOffset.Store(math.MaxInt64)
@@ -133,6 +136,7 @@ func TestWalTrimUpToCommitOffset(t *testing.T) {
 				slog.Info(
 					"checking...",
 					slog.Int64("first-offset", w.FirstOffset()),
+					slog.String("TestName", t.Name()),
 				)
 				return w.FirstOffset() == 2
 			}, 10*time.Second, 10*time.Millisecond)
@@ -151,6 +155,7 @@ func TestWalTrimUpToCommitOffset(t *testing.T) {
 				slog.Info(
 					"checking...",
 					slog.Int64("first-offset", w.FirstOffset()),
+					slog.String("TestName", t.Name()),
 				)
 				return w.FirstOffset() == 87
 			}, 10*time.Second, 10*time.Millisecond)