Repository: liftbridge-io/liftbridge
Pull request #65 (2019-06-25 05:31:11)

Clean up firstOffset/lastOffset logic
Remove TODOs and commit to using int64 for offsets.


Patch:
diff --git a/server/api.go b/server/api.go
index b374dfa..554731a 100644
--- a/server/api.go
+++ b/server/api.go
@@ -134,6 +134,11 @@ func (a *apiServer) subscribe(ctx context.Context, stream *stream,
 			fmt.Sprintf("Unknown StartPosition %s", req.StartPosition))
 	}
 
+	// If log is empty, next offset will be 0.
+	if startOffset < 0 {
+		startOffset = 0
+	}
+
 	var (
 		ch          = make(chan *client.Message)
 		errCh       = make(chan *status.Status)
diff --git a/server/commitlog.go b/server/commitlog.go
index 357d427..b31d968 100644
--- a/server/commitlog.go
+++ b/server/commitlog.go
@@ -19,10 +19,12 @@ type CommitLog interface {
 	// Truncate removes all messages from the log starting at the given offset.
 	Truncate(offset int64) error
 
-	// NewestOffset returns the offset of the last message in the log.
+	// NewestOffset returns the offset of the last message in the log or -1 if
+	// empty.
 	NewestOffset() int64
 
-	// OldestOffset returns the offset of the first message in the log.
+	// OldestOffset returns the offset of the first message in the log or -1 if
+	// empty.
 	OldestOffset() int64
 
 	// OffsetForTimestamp returns the earliest offset whose timestamp is
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index a7b2bb0..3b6e487 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -231,16 +231,14 @@ func (l *CommitLog) append(segment *Segment, ms []byte, entries []*Entry) ([]int
 	return offsets, nil
 }
 
-// NewestOffset returns the offset of the last message in the log.
-// TODO: This should use a bool to indicate empty (once we switch to unsigned
-// offsets).
+// NewestOffset returns the offset of the last message in the log or -1 if
+// empty.
 func (l *CommitLog) NewestOffset() int64 {
 	return l.activeSegment().NextOffset() - 1
 }
 
-// OldestOffset returns the offset of the first message in the log.
-// TODO: This should use a bool to indicate empty (once we switch to unsigned
-// offsets).
+// OldestOffset returns the offset of the first message in the log or -1 if
+// empty.
 func (l *CommitLog) OldestOffset() int64 {
 	l.mu.RLock()
 	defer l.mu.RUnlock()
diff --git a/server/commitlog/segment.go b/server/commitlog/segment.go
index 31df9ec..d6656ce 100644
--- a/server/commitlog/segment.go
+++ b/server/commitlog/segment.go
@@ -49,8 +49,8 @@ type Segment struct {
 	log            *os.File
 	Index          *Index
 	BaseOffset     int64
-	firstOffset    *int64
-	lastOffset     *int64
+	firstOffset    int64
+	lastOffset     int64
 	firstWriteTime int64
 	lastWriteTime  int64
 	position       int64
@@ -67,11 +67,13 @@ type Segment struct {
 
 func NewSegment(path string, baseOffset, maxBytes int64, isNew bool, suffix string) (*Segment, error) {
 	s := &Segment{
-		maxBytes:   maxBytes,
-		BaseOffset: baseOffset,
-		path:       path,
-		suffix:     suffix,
-		waiters:    make(map[contextReader]chan struct{}),
+		maxBytes:    maxBytes,
+		BaseOffset:  baseOffset,
+		firstOffset: -1,
+		lastOffset:  -1,
+		path:        path,
+		suffix:      suffix,
+		waiters:     make(map[contextReader]chan struct{}),
 	}
 	// If this is a new segment, ensure the file doesn't already exist.
 	if isNew && exists(s.logPath()) {
@@ -112,14 +114,14 @@ func (s *Segment) setupIndex() (err error) {
 	}
 	// If lastEntry is nil, the index is empty.
 	if lastEntry != nil {
-		s.lastOffset = &lastEntry.Offset
+		s.lastOffset = lastEntry.Offset
 		s.lastWriteTime = lastEntry.Timestamp
 		// Read the first entry to get firstOffset and firstWriteTime.
 		var firstEntry Entry
 		if err := s.Index.ReadEntryAtFileOffset(&firstEntry, 0); err != nil {
 			return err
 		}
-		s.firstOffset = &firstEntry.Offset
+		s.firstOffset = firstEntry.Offset
 		s.firstWriteTime = firstEntry.Timestamp
 	}
 	return nil
@@ -163,30 +165,22 @@ func (s *Segment) NextOffset() int64 {
 	defer s.RUnlock()
 	// If the segment hasn't been written to, the next offset should be the
 	// base offset.
-	if s.lastOffset == nil {
+	if s.lastOffset == -1 {
 		return s.BaseOffset
 	}
-	return *s.lastOffset + 1
+	return s.lastOffset + 1
 }
 
 func (s *Segment) FirstOffset() int64 {
 	s.RLock()
 	defer s.RUnlock()
-	// TODO: Should return a value and bool when we switch to unsigned ints?
-	if s.firstOffset == nil {
-		return 0
-	}
-	return *s.firstOffset
+	return s.firstOffset
 }
 
 func (s *Segment) LastOffset() int64 {
 	s.RLock()
 	defer s.RUnlock()
-	// TODO: Should return a value and bool when we switch to unsigned ints?
-	if s.lastOffset == nil {
-		return -1
-	}
-	return *s.lastOffset
+	return s.lastOffset
 }
 
 func (s *Segment) Position() int64 {
@@ -198,7 +192,7 @@ func (s *Segment) Position() int64 {
 func (s *Segment) IsEmpty() bool {
 	s.RLock()
 	defer s.RUnlock()
-	return s.firstOffset == nil
+	return s.firstOffset == -1
 }
 
 func (s *Segment) MessageCount() int64 {
@@ -229,11 +223,11 @@ func (s *Segment) write(p []byte, entries []*Entry) (n int, err error) {
 	s.position += int64(n)
 	if s.firstWriteTime == 0 {
 		first := entries[0]
-		s.firstOffset = &first.Offset
+		s.firstOffset = first.Offset
 		s.firstWriteTime = first.Timestamp
 	}
 	last := entries[len(entries)-1]
-	s.lastOffset = &last.Offset
+	s.lastOffset = last.Offset
 	s.lastWriteTime = last.Timestamp
 	s.notifyWaiters()
 	return n, nil


Test patch:
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index 80f01a8..506a910 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -139,7 +139,7 @@ func TestOffsets(t *testing.T) {
 	})
 	defer l.Close()
 	defer cleanup()
-	require.Equal(t, int64(0), l.OldestOffset())
+	require.Equal(t, int64(-1), l.OldestOffset())
 	require.Equal(t, int64(-1), l.NewestOffset())
 
 	numMsgs := 5
diff --git a/server/commitlog/compact_cleaner_test.go b/server/commitlog/compact_cleaner_test.go
index 8b12b98..e376944 100644
--- a/server/commitlog/compact_cleaner_test.go
+++ b/server/commitlog/compact_cleaner_test.go
@@ -234,7 +234,7 @@ func TestCompactCleanerTruncateConcurrent(t *testing.T) {
 	require.NoError(t, l.Clean())
 	require.NoError(t, l.Truncate(0))
 
-	require.Equal(t, int64(0), l.OldestOffset())
+	require.Equal(t, int64(-1), l.OldestOffset())
 }
 
 func BenchmarkClean1GBSegments(b *testing.B) {
diff --git a/server/commitlog/util_test.go b/server/commitlog/util_test.go
index f5db482..ee7cd88 100644
--- a/server/commitlog/util_test.go
+++ b/server/commitlog/util_test.go
@@ -8,11 +8,11 @@ import (
 
 func TestFindSegment(t *testing.T) {
 	segments := []*Segment{
-		&Segment{BaseOffset: 0, lastOffset: int64Ptr(9)},
-		&Segment{BaseOffset: 10, lastOffset: int64Ptr(19)},
-		&Segment{BaseOffset: 20, lastOffset: int64Ptr(29)},
-		&Segment{BaseOffset: 30, lastOffset: int64Ptr(39)},
-		&Segment{BaseOffset: 40, lastOffset: int64Ptr(49)},
+		&Segment{BaseOffset: 0, lastOffset: 9},
+		&Segment{BaseOffset: 10, lastOffset: 19},
+		&Segment{BaseOffset: 20, lastOffset: 29},
+		&Segment{BaseOffset: 30, lastOffset: 39},
+		&Segment{BaseOffset: 40, lastOffset: 49},
 	}
 	seg, idx := findSegment(segments, 0)
 	require.Equal(t, 0, idx)