Repository: liftbridge-io/liftbridge
Pull request #117 (2019-10-28 20:53:48)

Linter fixes and related code reorg


Patch:
diff --git a/go.sum b/go.sum
index 0d3843c..1bad24f 100644
--- a/go.sum
+++ b/go.sum
@@ -19,6 +19,7 @@ github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4=
 github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
 github.com/circonus-labs/circonus-gometrics v2.3.1+incompatible/go.mod h1:nmEj6Dob7S7YxXgwXpfOuvO54S+tGdZdw9fuRZt25Ag=
 github.com/circonus-labs/circonusllhist v0.1.3/go.mod h1:kMXHVDlOchFAehlya5ePtbp5jckzBHf4XRpQvBOLI+I=
+github.com/client9/misspell v0.3.4 h1:ta993UF76GwbvJcIo3Y68y/M3WxlpEHPWIGDkJYwzJI=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
diff --git a/main.go b/main.go
index 9fc80a2..5958550 100644
--- a/main.go
+++ b/main.go
@@ -22,70 +22,76 @@ func main() {
 	app.Usage = "Lightweight, fault-tolerant message streams"
 	app.Version = version
 	app.Flags = getFlags()
-	app.Action = func(c *cli.Context) error {
-		// Read config from file if present.
-		config, err := server.NewConfig(c.String("config"))
-		if err != nil {
-			return err
-		}
+	app.Action = start
+	if err := app.Run(os.Args); err != nil {
+		panic(err)
+	}
+}
 
-		// Override with flags.
-		if c.IsSet("id") {
-			config.Clustering.ServerID = c.String("id")
-		}
-		if c.IsSet("namespace") {
-			config.Clustering.Namespace = c.String("namespace")
-		}
-		if c.IsSet("port") {
-			config.Port = c.Int("port")
-		}
-		if c.IsSet("level") {
-			level, err := server.GetLogLevel(c.String("level"))
-			if err != nil {
-				return err
-			}
-			config.LogLevel = level
-		}
-		if c.IsSet("raft-bootstrap-seed") {
-			config.Clustering.RaftBootstrapSeed = c.Bool("raft-bootstrap-seed")
-		}
-		if c.IsSet("raft-bootstrap-peers") {
-			config.Clustering.RaftBootstrapPeers = c.StringSlice("raft-bootstrap-peers")
-		}
-		if c.IsSet("data-dir") {
-			config.DataDir = c.String("data-dir")
-		}
-		if c.IsSet("tls-cert") {
-			config.TLSCert = c.String("tls-cert")
-		}
-		if c.IsSet("tls-key") {
-			config.TLSKey = c.String("tls-key")
-		}
-		if c.IsSet("tls-client-auth") {
-			config.TLSClientAuth = c.Bool("tls-client-auth")
-		}
-		if c.IsSet("tls-client-auth-ca") {
-			config.TLSClientAuthCA = c.String("tls-client-auth-ca")
-		}
-		if c.IsSet("nats-servers") {
-			natsServers, err := normalizeNatsServers(c.StringSlice("nats-servers"))
-			if err != nil {
-				return err
-			}
-			config.NATS.Servers = natsServers
-		}
+func start(c *cli.Context) error {
+	// Read config from file if present.
+	config, err := server.NewConfig(c.String("config"))
+	if err != nil {
+		return err
+	}
+	if err := overrideFromFlags(c, config); err != nil {
+		return err
+	}
+	server := server.New(config)
+	if err := server.Start(); err != nil {
+		return err
+	}
+	runtime.Goexit()
+	return nil
+}
 
-		server := server.New(config)
-		if err := server.Start(); err != nil {
+func overrideFromFlags(c *cli.Context, config *server.Config) error {
+	// Override with flags.
+	if c.IsSet("id") {
+		config.Clustering.ServerID = c.String("id")
+	}
+	if c.IsSet("namespace") {
+		config.Clustering.Namespace = c.String("namespace")
+	}
+	if c.IsSet("port") {
+		config.Port = c.Int("port")
+	}
+	if c.IsSet("level") {
+		level, err := server.GetLogLevel(c.String("level"))
+		if err != nil {
 			return err
 		}
-		runtime.Goexit()
-		return nil
+		config.LogLevel = level
 	}
-
-	if err := app.Run(os.Args); err != nil {
-		panic(err)
+	if c.IsSet("raft-bootstrap-seed") {
+		config.Clustering.RaftBootstrapSeed = c.Bool("raft-bootstrap-seed")
+	}
+	if c.IsSet("raft-bootstrap-peers") {
+		config.Clustering.RaftBootstrapPeers = c.StringSlice("raft-bootstrap-peers")
+	}
+	if c.IsSet("data-dir") {
+		config.DataDir = c.String("data-dir")
+	}
+	if c.IsSet("tls-cert") {
+		config.TLSCert = c.String("tls-cert")
+	}
+	if c.IsSet("tls-key") {
+		config.TLSKey = c.String("tls-key")
+	}
+	if c.IsSet("tls-client-auth") {
+		config.TLSClientAuth = c.Bool("tls-client-auth")
+	}
+	if c.IsSet("tls-client-auth-ca") {
+		config.TLSClientAuthCA = c.String("tls-client-auth-ca")
+	}
+	if c.IsSet("nats-servers") {
+		natsServers, err := normalizeNatsServers(c.StringSlice("nats-servers"))
+		if err != nil {
+			return err
+		}
+		config.NATS.Servers = natsServers
 	}
+	return nil
 }
 
 func getFlags() []cli.Flag {
diff --git a/server/api.go b/server/api.go
index aa98d11..5e058c5 100644
--- a/server/api.go
+++ b/server/api.go
@@ -11,6 +11,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/liftbridge-io/liftbridge/server/commitlog"
 	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
@@ -216,32 +217,9 @@ func (a *apiServer) subscribe(ctx context.Context, partition *partition,
 	req *client.SubscribeRequest, cancel chan struct{}) (
 	<-chan *client.Message, <-chan *status.Status, *status.Status) {
 
-	var startOffset int64
-	switch req.StartPosition {
-	case client.StartPosition_OFFSET:
-		startOffset = req.StartOffset
-	case client.StartPosition_TIMESTAMP:
-		offset, err := partition.log.OffsetForTimestamp(req.StartTimestamp)
-		if err != nil {
-			return nil, nil, status.New(
-				codes.Internal, fmt.Sprintf("Failed to lookup offset for timestamp: %v", err))
-		}
-		startOffset = offset
-	case client.StartPosition_EARLIEST:
-		startOffset = partition.log.OldestOffset()
-	case client.StartPosition_LATEST:
-		startOffset = partition.log.NewestOffset()
-	case client.StartPosition_NEW_ONLY:
-		startOffset = partition.log.NewestOffset() + 1
-	default:
-		return nil, nil, status.New(
-			codes.InvalidArgument,
-			fmt.Sprintf("Unknown StartPosition %s", req.StartPosition))
-	}
-
-	// If log is empty, next offset will be 0.
-	if startOffset < 0 {
-		startOffset = 0
+	startOffset, st := getStartOffset(req, partition.log)
+	if st != nil {
+		return nil, nil, st
 	}
 
 	var (
@@ -288,3 +266,35 @@ func (a *apiServer) subscribe(ctx context.Context, partition *partition,
 
 	return ch, errCh, nil
 }
+
+func getStartOffset(req *client.SubscribeRequest, log commitlog.CommitLog) (int64, *status.Status) {
+	var startOffset int64
+	switch req.StartPosition {
+	case client.StartPosition_OFFSET:
+		startOffset = req.StartOffset
+	case client.StartPosition_TIMESTAMP:
+		offset, err := log.OffsetForTimestamp(req.StartTimestamp)
+		if err != nil {
+			return startOffset, status.New(
+				codes.Internal, fmt.Sprintf("Failed to lookup offset for timestamp: %v", err))
+		}
+		startOffset = offset
+	case client.StartPosition_EARLIEST:
+		startOffset = log.OldestOffset()
+	case client.StartPosition_LATEST:
+		startOffset = log.NewestOffset()
+	case client.StartPosition_NEW_ONLY:
+		startOffset = log.NewestOffset() + 1
+	default:
+		return startOffset, status.New(
+			codes.InvalidArgument,
+			fmt.Sprintf("Unknown StartPosition %s", req.StartPosition))
+	}
+
+	// If log is empty, next offset will be 0.
+	if startOffset < 0 {
+		startOffset = 0
+	}
+
+	return startOffset, nil
+}
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index f53f35a..e17c11a 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -32,25 +32,25 @@ const (
 	defaultCleanerInterval      = 5 * time.Minute
 )
 
-// CommitLog implements the server.CommitLog interface, which is a durable
-// write-ahead log.
-type CommitLog struct {
+// commitLog implements the CommitLog interface, which is a durable write-ahead
+// log.
+type commitLog struct {
 	Options
-	deleteCleaner    *DeleteCleaner
-	compactCleaner   *CompactCleaner
+	deleteCleaner    *deleteCleaner
+	compactCleaner   *compactCleaner
 	name             string
 	mu               sync.RWMutex
 	hw               int64
 	closed           chan struct{}
-	segments         []*Segment
-	vActiveSegment   *Segment
+	segments         []*segment
+	vActiveSegment   *segment
 	hwWaiters        map[contextReader]chan struct{}
 	leaderEpochCache *leaderEpochCache
 }
 
-// Options contains settings for configuring a CommitLog.
+// Options contains settings for configuring a commitLog.
 type Options struct {
-	Name                 string        // CommitLog name
+	Name                 string        // commitLog name
 	Path                 string        // Path to log directory
 	MaxSegmentBytes      int64         // Max bytes a Segment can contain before creating a new one
 	MaxLogBytes          int64         // Retention by bytes
@@ -66,7 +66,7 @@ type Options struct {
 
 // New creates a new CommitLog and starts a background goroutine which
 // periodically checkpoints the high watermark to disk.
-func New(opts Options) (*CommitLog, error) {
+func New(opts Options) (CommitLog, error) {
 	if opts.Path == "" {
 		return nil, errors.New("path is empty")
 	}
@@ -86,21 +86,21 @@ func New(opts Options) (*CommitLog, error) {
 		opts.CleanerInterval = defaultCleanerInterval
 	}
 
-	cleanerOpts := DeleteCleanerOptions{
+	cleanerOpts := deleteCleanerOptions{
 		Name:   opts.Path,
 		Logger: opts.Logger,
 	}
 	cleanerOpts.Retention.Bytes = opts.MaxLogBytes
 	cleanerOpts.Retention.Messages = opts.MaxLogMessages
 	cleanerOpts.Retention.Age = opts.MaxLogAge
-	cleaner := NewDeleteCleaner(cleanerOpts)
+	cleaner := newDeleteCleaner(cleanerOpts)
 
-	compactCleanerOpts := CompactCleanerOptions{
+	compactCleanerOpts := compactCleanerOptions{
 		Name:          opts.Name,
 		Logger:        opts.Logger,
 		MaxGoroutines: opts.CompactMaxGoroutines,
 	}
-	compactCleaner := NewCompactCleaner(compactCleanerOpts)
+	compactCleaner := newCompactCleaner(compactCleanerOpts)
 
 	path, _ := filepath.Abs(opts.Path)
 	epochCache, err := newLeaderEpochCache(opts.Name, path, opts.Logger)
@@ -108,7 +108,7 @@ func New(opts Options) (*CommitLog, error) {
 		return nil, err
 	}
 
-	l := &CommitLog{
+	l := &commitLog{
 		Options:          opts,
 		name:             filepath.Base(path),
 		deleteCleaner:    cleaner,
@@ -147,7 +147,7 @@ func New(opts Options) (*CommitLog, error) {
 	return l, nil
 }
 
-func (l *CommitLog) init() error {
+func (l *commitLog) init() error {
 	err := os.MkdirAll(l.Path, 0755)
 	if err != nil {
 		return errors.Wrap(err, "mkdir failed")
@@ -155,7 +155,7 @@ func (l *CommitLog) init() error {
 	return nil
 }
 
-func (l *CommitLog) open() error {
+func (l *commitLog) open() error {
 	files, err := ioutil.ReadDir(l.Path)
 	if err != nil {
 		return errors.Wrap(err, "read dir failed")
@@ -179,7 +179,7 @@ func (l *CommitLog) open() error {
 			if err != nil {
 				return err
 			}
-			segment, err := NewSegment(l.Path, int64(baseOffset), l.MaxSegmentBytes, false, "")
+			segment, err := newSegment(l.Path, int64(baseOffset), l.MaxSegmentBytes, false, "")
 			if err != nil {
 				return err
 			}
@@ -198,7 +198,7 @@ func (l *CommitLog) open() error {
 		}
 	}
 	if len(l.segments) == 0 {
-		segment, err := NewSegment(l.Path, 0, l.MaxSegmentBytes, true, "")
+		segment, err := newSegment(l.Path, 0, l.MaxSegmentBytes, true, "")
 		if err != nil {
 			return err
 		}
@@ -212,7 +212,7 @@ func (l *CommitLog) open() error {
 
 // Append writes the given batch of messages to the log and returns their
 // corresponding offsets in the log.
-func (l *CommitLog) Append(msgs []*proto.Message) ([]int64, error) {
+func (l *commitLog) Append(msgs []*proto.Message) ([]int64, error) {
 	if _, err := l.checkAndPerformSplit(); err != nil {
 		return nil, err
 	}
@@ -220,7 +220,7 @@ func (l *CommitLog) Append(msgs []*proto.Message) ([]int64, error) {
 		segment          = l.activeSegment()
 		basePosition     = segment.Position()
 		baseOffset       = segment.NextOffset()
-		ms, entries, err = NewMessageSetFromProto(baseOffset, basePosition, msgs)
+		ms, entries, err = newMessageSetFromProto(baseOffset, basePosition, msgs)
 	)
 	if err != nil {
 		return nil, err
@@ -230,19 +230,19 @@ func (l *CommitLog) Append(msgs []*proto.Message) ([]int64, error) {
 
 // AppendMessageSet writes the given message set data to the log and returns
 // the corresponding offsets in the log.
-func (l *CommitLog) AppendMessageSet(ms []byte) ([]int64, error) {
+func (l *commitLog) AppendMessageSet(ms []byte) ([]int64, error) {
 	if _, err := l.checkAndPerformSplit(); err != nil {
 		return nil, err
 	}
 	var (
 		segment      = l.activeSegment()
 		basePosition = segment.Position()
-		entries      = EntriesForMessageSet(basePosition, ms)
+		entries      = entriesForMessageSet(basePosition, ms)
 	)
 	return l.append(segment, ms, entries)
 }
 
-func (l *CommitLog) append(segment *Segment, ms []byte, entries []*Entry) ([]int64, error) {
+func (l *commitLog) append(segment *segment, ms []byte, entries []*entry) ([]int64, error) {
 	if err := segment.WriteMessageSet(ms, entries); err != nil {
 		return nil, err
 	}
@@ -266,13 +266,13 @@ func (l *CommitLog) append(segment *Segment, ms []byte, entries []*Entry) ([]int
 
 // NewestOffset returns the offset of the last message in the log or -1 if
 // empty.
-func (l *CommitLog) NewestOffset() int64 {
+func (l *commitLog) NewestOffset() int64 {
 	return l.activeSegment().NextOffset() - 1
 }
 
 // OldestOffset returns the offset of the first message in the log or -1 if
 // empty.
-func (l *CommitLog) OldestOffset() int64 {
+func (l *commitLog) OldestOffset() int64 {
 	l.mu.RLock()
 	defer l.mu.RUnlock()
 	return l.segments[0].FirstOffset()
@@ -280,7 +280,7 @@ func (l *CommitLog) OldestOffset() int64 {
 
 // OffsetForTimestamp returns the earliest offset whose timestamp is greater
 // than or equal to the given timestamp.
-func (l *CommitLog) OffsetForTimestamp(timestamp int64) (int64, error) {
+func (l *commitLog) OffsetForTimestamp(timestamp int64) (int64, error) {
 	l.mu.RLock()
 	defer l.mu.RUnlock()
 
@@ -293,7 +293,7 @@ func (l *CommitLog) OffsetForTimestamp(timestamp int64) (int64, error) {
 	// Search the previous segment for the first entry whose timestamp is
 	// greater than or equal to the given timestamp. If this is the first
 	// segment, just search it.
-	var seg *Segment
+	var seg *segment
 	if idx == 0 {
 		seg = l.segments[0]
 	} else {
@@ -323,7 +323,7 @@ func (l *CommitLog) OffsetForTimestamp(timestamp int64) (int64, error) {
 
 // SetHighWatermark sets the high watermark on the log. All messages up to and
 // including the high watermark are considered committed.
-func (l *CommitLog) SetHighWatermark(hw int64) {
+func (l *commitLog) SetHighWatermark(hw int64) {
 	l.mu.Lock()
 	if hw > l.hw {
 		l.hw = hw
@@ -336,21 +336,21 @@ func (l *CommitLog) SetHighWatermark(hw int64) {
 // OverrideHighWatermark sets the high watermark on the log using the given
 // value, even if the value is less than the current HW. This is used for unit
 // testing purposes.
-func (l *CommitLog) OverrideHighWatermark(hw int64) {
+func (l *commitLog) OverrideHighWatermark(hw int64) {
 	l.mu.Lock()
 	l.hw = hw
 	l.notifyHWWaiters()
 	l.mu.Unlock()
 }
 
-func (l *CommitLog) notifyHWWaiters() {
+func (l *commitLog) notifyHWWaiters() {
 	for r, ch := range l.hwWaiters {
 		close(ch)
 		delete(l.hwWaiters, r)
 	}
 }
 
-func (l *CommitLog) waitForHW(r contextReader, hw int64) <-chan struct{} {
+func (l *commitLog) waitForHW(r contextReader, hw int64) <-chan struct{} {
 	wait := make(chan struct{})
 	l.mu.Lock()
 	// Check if HW has changed.
@@ -363,28 +363,28 @@ func (l *CommitLog) waitForHW(r contextReader, hw int64) <-chan struct{} {
 	return wait
 }
 
-func (l *CommitLog) removeHWWaiter(r contextReader) {
+func (l *commitLog) removeHWWaiter(r contextReader) {
 	l.mu.Lock()
 	delete(l.hwWaiters, r)
 	l.mu.Unlock()
 }
 
 // HighWatermark returns the high watermark for the log.
-func (l *CommitLog) HighWatermark() int64 {
+func (l *commitLog) HighWatermark() int64 {
 	l.mu.RLock()
 	defer l.mu.RUnlock()
 	return l.hw
 }
 
 // NewLeaderEpoch indicates the log is entering a new leader epoch.
-func (l *CommitLog) NewLeaderEpoch(epoch uint64) error {
+func (l *commitLog) NewLeaderEpoch(epoch uint64) error {
 	return l.leaderEpochCache.Assign(epoch, l.NewestOffset())
 }
 
 // LastOffsetForLeaderEpoch returns the start offset of the first leader epoch
 // larger than the provided one or the log end offset if the current epoch
 // equals the provided one.
-func (l *CommitLog) LastOffsetForLeaderEpoch(epoch uint64) int64 {
+func (l *commitLog) LastOffsetForLeaderEpoch(epoch uint64) int64 {
 	offset := l.leaderEpochCache.LastOffsetForLeaderEpoch(epoch)
 	if offset == -1 {
 		offset = l.activeSegment().NextOffset() - 1
@@ -393,17 +393,17 @@ func (l *CommitLog) LastOffsetForLeaderEpoch(epoch uint64) int64 {
 }
 
 // LastLeaderEpoch returns the latest leader epoch for the log.
-func (l *CommitLog) LastLeaderEpoch() uint64 {
+func (l *commitLog) LastLeaderEpoch() uint64 {
 	return l.leaderEpochCache.LastLeaderEpoch()
 }
 
-func (l *CommitLog) activeSegment() *Segment {
-	return (*Segment)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&l.vActiveSegment))))
+func (l *commitLog) activeSegment() *segment {
+	return (*segment)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&l.vActiveSegment))))
 }
 
 // Close closes each log segment file and stops the background goroutine
 // checkpointing the high watermark to disk.
-func (l *CommitLog) Close() error {
+func (l *commitLog) Close() error {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 	if err := l.checkpointHW(); err != nil {
@@ -420,7 +420,7 @@ func (l *CommitLog) Close() error {
 
 // Delete closes the log and removes all data associated with it from the
 // filesystem.
-func (l *CommitLog) Delete() error {
+func (l *commitLog) Delete() error {
 	if err := l.Close(); err != nil {
 		return err
 	}
@@ -428,11 +428,11 @@ func (l *CommitLog) Delete() error {
 }
 
 // Truncate removes all messages from the log starting at the given offset.
-func (l *CommitLog) Truncate(offset int64) error {
+func (l *commitLog) Truncate(offset int64) error {
 	l.mu.Lock()
 	defer l.mu.Unlock()
-	segment, idx := findSegment(l.segments, offset)
-	if segment == nil {
+	seg, idx := findSegment(l.segments, offset)
+	if seg == nil {
 		// Nothing to truncate.
 		return nil
 	}
@@ -450,11 +450,11 @@ func (l *CommitLog) Truncate(offset int64) error {
 
 	// Delete the segment if its base offset is the target offset, provided
 	// it's not the first segment.
-	if segment.BaseOffset == offset {
+	if seg.BaseOffset == offset {
 		if idx == 0 {
 			replace = true
 		} else {
-			if err := segment.Delete(); err != nil {
+			if err := seg.Delete(); err != nil {
 				return err
 			}
 			deleted++
@@ -464,7 +464,7 @@ func (l *CommitLog) Truncate(offset int64) error {
 	}
 
 	// Retain all preceding segments.
-	segments := make([]*Segment, len(l.segments)-deleted)
+	segments := make([]*segment, len(l.segments)-deleted)
 	for i := 0; i < idx; i++ {
 		segments[i] = l.segments[i]
 	}
@@ -472,22 +472,22 @@ func (l *CommitLog) Truncate(offset int64) error {
 	// Replace segment containing offset with truncated segment.
 	if replace {
 		var (
-			ss              = NewSegmentScanner(segment)
-			newSegment, err = segment.Truncated()
+			ss              = newSegmentScanner(seg)
+			newSegment, err = seg.Truncated()
 		)
 		if err != nil {
 			return err
 		}
-		for ms, entry, err := ss.Scan(); err == nil; ms, entry, err = ss.Scan() {
+		for ms, e, err := ss.Scan(); err == nil; ms, e, err = ss.Scan() {
 			if ms.Offset() < offset {
-				if err := newSegment.WriteMessageSet(ms, []*Entry{entry}); err != nil {
+				if err := newSegment.WriteMessageSet(ms, []*entry{e}); err != nil {
 					return err
 				}
 			} else {
 				break
 			}
 		}
-		if err = newSegment.Replace(segment); err != nil {
+		if err = newSegment.Replace(seg); err != nil {
 			return err
 		}
 		segments[idx] = newSegment
@@ -499,7 +499,7 @@ func (l *CommitLog) Truncate(offset int64) error {
 	return l.leaderEpochCache.ClearLatest(offset)
 }
 
-func (l *CommitLog) Segments() []*Segment {
+func (l *commitLog) Segments() []*segment {
 	l.mu.RLock()
 	defer l.mu.RUnlock()
 	return l.segments
@@ -509,7 +509,7 @@ func (l *CommitLog) Segments() []*Segment {
 // the given log end offset are added to the log. If the given offset is no
 // longer the log end offset, the channel is closed immediately. Waiter is an
 // opaque value that uniquely identifies the entity waiting for data.
-func (l *CommitLog) NotifyLEO(waiter interface{}, leo int64) <-chan struct{} {
+func (l *commitLog) NotifyLEO(waiter interface{}, leo int64) <-chan struct{} {
 	return l.activeSegment().WaitForLEO(waiter, leo)
 }
 
@@ -518,7 +518,7 @@ func (l *CommitLog) NotifyLEO(waiter interface{}, leo int64) <-chan struct{} {
 // the first message was written to it. It then performs the split if eligible,
 // returning any error resulting from the split. The returned bool indicates if
 // a split was performed.
-func (l *CommitLog) checkAndPerformSplit() (bool, error) {
+func (l *commitLog) checkAndPerformSplit() (bool, error) {
 	// Do this in a loop because segment splitting may fail due to a competing
 	// thread performing the split at the same time. If this happens, we just
 	// retry the check on the new active segment.
@@ -541,10 +541,10 @@ func (l *CommitLog) checkAndPerformSplit() (bool, error) {
 	}
 }
 
-func (l *CommitLog) split(oldActiveSegment *Segment) error {
+func (l *commitLog) split(oldActiveSegment *segment) error {
 	offset := l.NewestOffset() + 1
 	l.Logger.Debugf("Appending new log segment for %s with base offset %d", l.Path, offset)
-	segment, err := NewSegment(l.Path, offset, l.MaxSegmentBytes, true, "")
+	segment, err := newSegment(l.Path, offset, l.MaxSegmentBytes, true, "")
 	if err != nil {
 		return err
 	}
@@ -564,7 +564,7 @@ func (l *CommitLog) split(oldActiveSegment *Segment) error {
 	return nil
 }
 
-func (l *CommitLog) cleanerLoop() {
+func (l *commitLog) cleanerLoop() {
 	ticker := time.NewTicker(l.CleanerInterval)
 	defer ticker.Stop()
 	for {
@@ -594,7 +594,7 @@ func (l *CommitLog) cleanerLoop() {
 }
 
 // Clean applies retention and compaction rules against the log, if applicable.
-func (l *CommitLog) Clean() error {
+func (l *commitLog) Clean() error {
 	l.mu.RLock()
 	oldSegments := l.segments
 	l.mu.RUnlock()
@@ -625,7 +625,7 @@ func (l *CommitLog) Clean() error {
 
 // rebaseSegments adds the segments in from to the end of the slice of segments
 // in to and adds any leader epoch offsets to the given leaderEpochCache.
-func (l *CommitLog) rebaseSegments(from, to []*Segment, epochCache *leaderEpochCache) []*Segment {
+func (l *commitLog) rebaseSegments(from, to []*segment, epochCache *leaderEpochCache) []*segment {
 	to = append(to, from...)
 	// Rebase any leader epoch offsets also. We don't check the error returned
 	// here because Rebase can't return an error since epochCache is not
@@ -637,7 +637,7 @@ func (l *CommitLog) rebaseSegments(from, to []*Segment, epochCache *leaderEpochC
 // clean returns the cleaned segments and, if compaction ran, a
 // *leaderEpochCache maintaining the start offset for each new leader epoch. If
 // compaction did not run, the leaderEpochCache will be nil.
-func (l *CommitLog) clean(segments []*Segment) ([]*Segment, *leaderEpochCache, error) {
+func (l *commitLog) clean(segments []*segment) ([]*segment, *leaderEpochCache, error) {
 	cleaned, err := l.deleteCleaner.Clean(segments)
 	if err != nil {
 		return nil, nil, err
@@ -652,7 +652,7 @@ func (l *CommitLog) clean(segments []*Segment) ([]*Segment, *leaderEpochCache, e
 	return cleaned, epochCache, nil
 }
 
-func (l *CommitLog) checkpointHWLoop() {
+func (l *commitLog) checkpointHWLoop() {
 	ticker := time.NewTicker(l.HWCheckpointInterval)
 	defer ticker.Stop()
 	for {
@@ -669,7 +669,7 @@ func (l *CommitLog) checkpointHWLoop() {
 	}
 }
 
-func (l *CommitLog) checkpointHW() error {
+func (l *commitLog) checkpointHW() error {
 	var (
 		hw   = l.hw
 		r    = strings.NewReader(strconv.FormatInt(hw, 10))
diff --git a/server/commitlog/compact_cleaner.go b/server/commitlog/compact_cleaner.go
index 53386a9..1451260 100644
--- a/server/commitlog/compact_cleaner.go
+++ b/server/commitlog/compact_cleaner.go
@@ -11,28 +11,28 @@ import (
 
 const defaultCompactMaxGoroutines = 10
 
-// CompactCleanerOptions contains configuration settings for the
-// CompactCleaner.
-type CompactCleanerOptions struct {
+// compactCleanerOptions contains configuration settings for the
+// compactCleaner.
+type compactCleanerOptions struct {
 	Logger        logger.Logger
 	Name          string
 	MaxGoroutines int
 }
 
-// CompactCleaner implements the compaction policy which replaces segments with
+// compactCleaner implements the compaction policy which replaces segments with
 // compacted ones, i.e. retaining only the last message for a given key.
-type CompactCleaner struct {
-	CompactCleanerOptions
+type compactCleaner struct {
+	compactCleanerOptions
 }
 
-// NewCompactCleaner returns a new Cleaner which performs log compaction by
+// NewCompactCleaner returns a new cleaner which performs log compaction by
 // rewriting segments such that they contain only the last message for a given
 // key.
-func NewCompactCleaner(opts CompactCleanerOptions) *CompactCleaner {
+func newCompactCleaner(opts compactCleanerOptions) *compactCleaner {
 	if opts.MaxGoroutines == 0 {
 		opts.MaxGoroutines = defaultCompactMaxGoroutines
 	}
-	return &CompactCleaner{opts}
+	return &compactCleaner{opts}
 }
 
 // Compact performs log compaction by rewriting segments such that they contain
@@ -41,7 +41,7 @@ func NewCompactCleaner(opts CompactCleanerOptions) *CompactCleaner {
 // comes first. This returns the compacted segments and a leaderEpochCache
 // containing the earliest offsets for each leader epoch or nil if nothing was
 // compacted.
-func (c *CompactCleaner) Compact(hw int64, segments []*Segment) ([]*Segment,
+func (c *compactCleaner) Compact(hw int64, segments []*segment) ([]*segment,
 	*leaderEpochCache, error) {
 
 	if len(segments) <= 1 {
@@ -82,14 +82,14 @@ func (k *keyOffset) get() int64 {
 	return k.offset
 }
 
-func (c *CompactCleaner) compact(hw int64, segments []*Segment) ([]*Segment,
+func (c *compactCleaner) compact(hw int64, segments []*segment) ([]*segment,
 	*leaderEpochCache, int, error) {
 
 	// Compact messages up to the last segment or HW, whichever is first, by
 	// scanning keys and retaining only the latest.
 	// TODO: Implement option for configuring minimum compaction lag.
 	var (
-		compacted  = make([]*Segment, 0, len(segments))
+		compacted  = make([]*segment, 0, len(segments))
 		epochCache = newLeaderEpochCacheNoFile(c.Name, c.Logger)
 		removed    = 0
 		keyOffsets = c.scanKeys(hw, segments)
@@ -98,53 +98,14 @@ func (c *CompactCleaner) compact(hw int64, segments []*Segment) ([]*Segment,
 	// Write new segments. Skip the last segment since we will not compact it.
 	// TODO: Join segments that are below the bytes limit.
 	for _, seg := range segments[:len(segments)-1] {
-		cleaned, err := seg.Cleaned()
+		cleaned, msgsRemoved, err := c.cleanSegment(seg, keyOffsets, hw, epochCache)
 		if err != nil {
 			return nil, nil, 0, err
 		}
-		ss := NewSegmentScanner(seg)
-		for ms, _, err := ss.Scan(); err == nil; ms, _, err = ss.Scan() {
-			var (
-				offset       = ms.Offset()
-				key          = ms.Message().Key()
-				leaderEpoch  = ms.LeaderEpoch()
-				latest, ok   = keyOffsets.Load(string(key))
-				latestOffset int64
-			)
-			if ok {
-				latestOffset = latest.(*keyOffset).get()
-			}
-
-			// Retain all messages with no keys and last message for each key.
-			// Also retain all messages after the HW.
-			if key == nil || offset == latestOffset || offset >= hw {
-				entries := EntriesForMessageSet(cleaned.Position(), ms)
-				if err := cleaned.WriteMessageSet(ms, entries); err != nil {
-					return nil, nil, 0, err
-				}
-				// Maintain start offset for each new leader epoch.
-				if leaderEpoch > epochCache.LastLeaderEpoch() {
-					if err := epochCache.Assign(leaderEpoch, offset); err != nil {
-						return nil, nil, 0, err
-					}
-				}
-			} else {
-				removed++
-			}
-		}
-
-		if cleaned.IsEmpty() {
-			// If the new segment is empty, remove it along with the old one.
-			if err := cleanupEmptySegment(cleaned, seg); err != nil {
-				return nil, nil, 0, err
-			}
-		} else {
-			// Otherwise replace the old segment with the compacted one.
-			if err = cleaned.Replace(seg); err != nil {
-				return nil, nil, 0, err
-			}
+		if cleaned != nil {
 			compacted = append(compacted, cleaned)
 		}
+		removed += msgsRemoved
 	}
 
 	// Add the last segment back in to the compacted list.
@@ -152,7 +113,7 @@ func (c *CompactCleaner) compact(hw int64, segments []*Segment) ([]*Segment,
 	compacted = append(compacted, last)
 
 	// Maintain start offset for each new leader epoch for the last segment.
-	ss := NewSegmentScanner(last)
+	ss := newSegmentScanner(last)
 	for ms, _, err := ss.Scan(); err == nil; ms, _, err = ss.Scan() {
 		leaderEpoch := ms.LeaderEpoch()
 		if leaderEpoch > epochCache.LastLeaderEpoch() {
@@ -165,12 +126,64 @@ func (c *CompactCleaner) compact(hw int64, segments []*Segment) ([]*Segment,
 	return compacted, epochCache, removed, nil
 }
 
-func (c *CompactCleaner) scanKeys(hw int64, segments []*Segment) *sync.Map {
+func (c *compactCleaner) cleanSegment(seg *segment, keyOffsets *sync.Map, hw int64,
+	epochCache *leaderEpochCache) (*segment, int, error) {
+
+	cleaned, err := seg.Cleaned()
+	if err != nil {
+		return nil, 0, err
+	}
+	var (
+		ss      = newSegmentScanner(seg)
+		removed = 0
+	)
+	for ms, _, err := ss.Scan(); err == nil; ms, _, err = ss.Scan() {
+		var (
+			offset       = ms.Offset()
+			key          = ms.Message().Key()
+			leaderEpoch  = ms.LeaderEpoch()
+			latest, ok   = keyOffsets.Load(string(key))
+			latestOffset int64
+		)
+		if ok {
+			latestOffset = latest.(*keyOffset).get()
+		}
+
+		// Retain all messages with no keys and last message for each key.
+		// Also retain all messages after the HW.
+		if key == nil || offset == latestOffset || offset >= hw {
+			entries := entriesForMessageSet(cleaned.Position(), ms)
+			if err := cleaned.WriteMessageSet(ms, entries); err != nil {
+				return nil, removed, err
+			}
+			// Maintain start offset for each new leader epoch.
+			if leaderEpoch > epochCache.LastLeaderEpoch() {
+				if err := epochCache.Assign(leaderEpoch, offset); err != nil {
+					return nil, removed, err
+				}
+			}
+		} else {
+			removed++
+		}
+	}
+
+	if cleaned.IsEmpty() {
+		// If the new segment is empty, remove it along with the old one.
+		return nil, removed, cleanupEmptySegment(cleaned, seg)
+	}
+	// Otherwise replace the old segment with the compacted one.
+	if err = cleaned.Replace(seg); err != nil {
+		return nil, removed, err
+	}
+	return cleaned, removed, nil
+}
+
+func (c *compactCleaner) scanKeys(hw int64, segments []*segment) *sync.Map {
 	var (
 		wg            sync.WaitGroup
 		keyOffsets    = new(sync.Map)
 		numGoroutines = c.MaxGoroutines
-		segmentC      = make(chan *Segment, len(segments))
+		segmentC      = make(chan *segment, len(segments))
 	)
 	if len(segments) < numGoroutines {
 		numGoroutines = len(segments)
@@ -190,10 +203,10 @@ func (c *CompactCleaner) scanKeys(hw int64, segments []*Segment) *sync.Map {
 	return keyOffsets
 }
 
-func (c *CompactCleaner) scanSegments(hw int64, ch <-chan *Segment, wg *sync.WaitGroup, keyOffsets *sync.Map) {
+func (c *compactCleaner) scanSegments(hw int64, ch <-chan *segment, wg *sync.WaitGroup, keyOffsets *sync.Map) {
 LOOP:
 	for seg := range ch {
-		ss := NewSegmentScanner(seg)
+		ss := newSegmentScanner(seg)
 		for ms, _, err := ss.Scan(); err == nil; ms, _, err = ss.Scan() {
 			offset := ms.Offset()
 			if offset > hw {
@@ -209,7 +222,7 @@ LOOP:
 	wg.Done()
 }
 
-func cleanupEmptySegment(new, old *Segment) error {
+func cleanupEmptySegment(new, old *segment) error {
 	// Delete the new segment if it's empty.
 	if err := new.Delete(); err != nil {
 		return err
diff --git a/server/commitlog/delete_cleaner.go b/server/commitlog/delete_cleaner.go
index 63130e5..a235882 100644
--- a/server/commitlog/delete_cleaner.go
+++ b/server/commitlog/delete_cleaner.go
@@ -14,8 +14,8 @@ var computeTTL = func(age time.Duration) int64 {
 	return time.Now().Add(-age).UnixNano()
 }
 
-// DeleteCleanerOptions contains configuration settings for the DeleteCleaner.
-type DeleteCleanerOptions struct {
+// deleteCleanerOptions contains configuration settings for the DeleteCleaner.
+type deleteCleanerOptions struct {
 	Retention struct {
 		Bytes    int64
 		Messages int64
@@ -25,21 +25,21 @@ type DeleteCleanerOptions struct {
 	Name   string
 }
 
-// DeleteCleaner implements the delete cleanup policy which deletes old log
+// deleteCleaner implements the delete cleanup policy which deletes old log
 // segments based on the retention policy.
-type DeleteCleaner struct {
-	DeleteCleanerOptions
+type deleteCleaner struct {
+	deleteCleanerOptions
 }
 
-// NewDeleteCleaner returns a new Cleaner which enforces log retention
+// newDeleteCleaner returns a new cleaner which enforces log retention
 // policies by deleting segments.
-func NewDeleteCleaner(opts DeleteCleanerOptions) *DeleteCleaner {
-	return &DeleteCleaner{opts}
+func newDeleteCleaner(opts deleteCleanerOptions) *deleteCleaner {
+	return &deleteCleaner{opts}
 }
 
 // Clean will enforce the log retention policy by deleting old segments.
 // Deletion only occurs at the segment granularity.
-func (c *DeleteCleaner) Clean(segments []*Segment) ([]*Segment, error) {
+func (c *deleteCleaner) Clean(segments []*segment) ([]*segment, error) {
 	var err error
 	if len(segments) == 0 || c.noRetentionLimits() {
 		return segments, nil
@@ -75,11 +75,11 @@ func (c *DeleteCleaner) Clean(segments []*Segment) ([]*Segment, error) {
 	return segments, nil
 }
 
-func (c DeleteCleaner) noRetentionLimits() bool {
+func (c *deleteCleaner) noRetentionLimits() bool {
 	return c.Retention.Bytes == 0 && c.Retention.Messages == 0 && c.Retention.Age == 0
 }
 
-func (c DeleteCleaner) applyMessagesLimit(segments []*Segment) ([]*Segment, error) {
+func (c *deleteCleaner) applyMessagesLimit(segments []*segment) ([]*segment, error) {
 	if len(segments) <= 1 {
 		return segments, nil
 	}
@@ -87,7 +87,7 @@ func (c DeleteCleaner) applyMessagesLimit(segments []*Segment) ([]*Segment, erro
 	// We start at the most recent segment and work our way backwards until we
 	// meet the retention size.
 	var (
-		cleanedSegments = make([]*Segment, 0, len(segments))
+		cleanedSegments = make([]*segment, 0, len(segments))
 		totalMessages   int64
 	)
 
@@ -98,7 +98,7 @@ func (c DeleteCleaner) applyMessagesLimit(segments []*Segment) ([]*Segment, erro
 		if totalMessages > c.Retention.Messages {
 			break
 		}
-		cleanedSegments = append([]*Segment{s}, cleanedSegments...)
+		cleanedSegments = append([]*segment{s}, cleanedSegments...)
 	}
 	if i > -1 {
 		for ; i > -1; i-- {
@@ -115,12 +115,12 @@ func (c DeleteCleaner) applyMessagesLimit(segments []*Segment) ([]*Segment, erro
 	return cleanedSegments, nil
 }
 
-func (c *DeleteCleaner) applyBytesLimit(segments []*Segment) ([]*Segment, error) {
+func (c *deleteCleaner) applyBytesLimit(segments []*segment) ([]*segment, error) {
 	// We start at the most recent segment and work our way backwards until we
 	// meet the retention size.
 	var (
 		lastSeg         = segments[len(segments)-1]
-		cleanedSegments = []*Segment{lastSeg}
+		cleanedSegments = []*segment{lastSeg}
 		totalBytes      = lastSeg.Position()
 	)
 
@@ -132,7 +132,7 @@ func (c *DeleteCleaner) applyBytesLimit(segments []*Segment) ([]*Segment, error)
 			if totalBytes > c.Retention.Bytes {
 				break
 			}
-			cleanedSegments = append([]*Segment{s}, cleanedSegments...)
+			cleanedSegments = append([]*segment{s}, cleanedSegments...)
 		}
 		if i > -1 {
 			for ; i > -1; i-- {
@@ -151,7 +151,7 @@ func (c *DeleteCleaner) applyBytesLimit(segments []*Segment) ([]*Segment, error)
 	return cleanedSegments, nil
 }
 
-func (c *DeleteCleaner) applyAgeLimit(segments []*Segment) ([]*Segment, error) {
+func (c *deleteCleaner) applyAgeLimit(segments []*segment) ([]*segment, error) {
 	// We must retain at least the active segment.
 	if len(segments) == 1 {
 		return segments, nil
diff --git a/server/commitlog/index.go b/server/commitlog/index.go
index 70def6b..00a5cb3 100644
--- a/server/commitlog/index.go
+++ b/server/commitlog/index.go
@@ -14,9 +14,7 @@ import (
 	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
-var (
-	ErrIndexCorrupt = errors.New("corrupt index file")
-)
+var errIndexCorrupt = errors.New("corrupt index file")
 
 const (
 	offsetWidth    = 4
@@ -26,7 +24,7 @@ const (
 	entryWidth     = offsetWidth + timestampWidth + positionWidth + sizeWidth
 )
 
-type Index struct {
+type index struct {
 	options
 	mmap     gommap.MMap
 	file     *os.File
@@ -35,7 +33,7 @@ type Index struct {
 	position int64
 }
 
-type Entry struct {
+type entry struct {
 	Offset      int64
 	Timestamp   int64
 	LeaderEpoch uint64
@@ -51,7 +49,7 @@ type relEntry struct {
 	Size      int32
 }
 
-func newRelEntry(e *Entry, baseOffset int64) relEntry {
+func newRelEntry(e *entry, baseOffset int64) relEntry {
 	return relEntry{
 		Offset:    int32(e.Offset - baseOffset),
 		Timestamp: e.Timestamp,
@@ -60,7 +58,7 @@ func newRelEntry(e *Entry, baseOffset int64) relEntry {
 	}
 }
 
-func (rel relEntry) fill(e *Entry, baseOffset int64) {
+func (rel relEntry) fill(e *entry, baseOffset int64) {
 	e.Offset = baseOffset + int64(rel.Offset)
 	e.Timestamp = rel.Timestamp
 	e.Position = int64(rel.Position)
@@ -73,14 +71,14 @@ type options struct {
 	baseOffset int64
 }
 
-func NewIndex(opts options) (idx *Index, err error) {
+func newIndex(opts options) (idx *index, err error) {
 	if opts.bytes == 0 {
 		opts.bytes = 10 * 1024 * 1024
 	}
 	if opts.path == "" {
 		return nil, errors.New("path is empty")
 	}
-	idx = &Index{
+	idx = &index{
 		options: opts,
 	}
 	idx.file, err = os.OpenFile(opts.path, os.O_RDWR|os.O_CREATE, 0666)
@@ -114,19 +112,19 @@ func NewIndex(opts options) (idx *Index, err error) {
 
 // Position returns the current position in the index to write to next. This
 // value also represents the total length of the index.
-func (idx *Index) Position() int64 {
+func (idx *index) Position() int64 {
 	idx.mu.RLock()
 	defer idx.mu.RUnlock()
 	return idx.position
 }
 
-func (idx *Index) CountEntries() int64 {
+func (idx *index) CountEntries() int64 {
 	idx.mu.RLock()
 	defer idx.mu.RUnlock()
 	return idx.position / entryWidth
 }
 
-func (idx *Index) writeEntries(entries []*Entry) (err error) {
+func (idx *index) writeEntries(entries []*entry) (err error) {
 	b := new(bytes.Buffer)
 	for _, entry := range entries {
 		relEntry := newRelEntry(entry, idx.baseOffset)
@@ -141,10 +139,10 @@ func (idx *Index) writeEntries(entries []*Entry) (err error) {
 	return nil
 }
 
-// ReadEntryAtFileOffset is used to read an Index entry at the given
-// byte offset of the Index file. ReadEntryAtLogOffset is generally
+// ReadEntryAtFileOffset is used to read an index entry at the given
+// byte offset of the index file. ReadEntryAtLogOffset is generally
 // more useful for higher level use.
-func (idx *Index) ReadEntryAtFileOffset(e *Entry, fileOffset int64) (err error) {
+func (idx *index) ReadEntryAtFileOffset(e *entry, fileOffset int64) (err error) {
 	p := make([]byte, entryWidth)
 	if _, err = idx.ReadAt(p, fileOffset); err != nil {
 		return err
@@ -161,13 +159,13 @@ func (idx *Index) ReadEntryAtFileOffset(e *Entry, fileOffset int64) (err error)
 	return nil
 }
 
-// ReadEntryAtLogOffset is used to read an Index entry at the given
-// log offset of the Index file.
-func (idx *Index) ReadEntryAtLogOffset(e *Entry, logOffset int64) error {
+// ReadEntryAtLogOffset is used to read an index entry at the given
+// log offset of the index file.
+func (idx *index) ReadEntryAtLogOffset(e *entry, logOffset int64) error {
 	return idx.ReadEntryAtFileOffset(e, logOffset*entryWidth)
 }
 
-func (idx *Index) ReadAt(p []byte, offset int64) (n int, err error) {
+func (idx *index) ReadAt(p []byte, offset int64) (n int, err error) {
 	idx.mu.RLock()
 	defer idx.mu.RUnlock()
 	if idx.position < offset+entryWidth {
@@ -177,7 +175,7 @@ func (idx *Index) ReadAt(p []byte, offset int64) (n int, err error) {
 	return n, nil
 }
 
-func (idx *Index) writeAt(p []byte, offset int64) (n int) {
+func (idx *index) writeAt(p []byte, offset int64) (n int) {
 	// Check if we need to expand the index file.
 	if pSize := int64(len(p)); offset+pSize >= idx.size {
 		// Expand the index file.
@@ -201,7 +199,7 @@ func (idx *Index) writeAt(p []byte, offset int64) (n int) {
 	return copy(idx.mmap[offset:], p)
 }
 
-func (idx *Index) Sync() error {
+func (idx *index) Sync() error {
 	idx.mu.Lock()
 	defer idx.mu.Unlock()
 	if err := idx.file.Sync(); err != nil {
@@ -213,7 +211,7 @@ func (idx *Index) Sync() error {
 	return nil
 }
 
-func (idx *Index) Close() error {
+func (idx *index) Close() error {
 	if err := idx.Sync(); err != nil {
 		return err
 	}
@@ -224,17 +222,17 @@ func (idx *Index) Close() error {
 }
 
 // Shrink truncates the memory-mapped index file to the size of its contents.
-func (idx *Index) Shrink() error {
+func (idx *index) Shrink() error {
 	idx.mu.RLock()
 	defer idx.mu.RUnlock()
 	return idx.file.Truncate(idx.position)
 }
 
-func (idx *Index) Name() string {
+func (idx *index) Name() string {
 	return idx.file.Name()
 }
 
-func (idx *Index) TruncateEntries(number int) error {
+func (idx *index) TruncateEntries(number int) error {
 	idx.mu.Lock()
 	defer idx.mu.Unlock()
 	if int64(number*entryWidth) > idx.position {
@@ -244,10 +242,10 @@ func (idx *Index) TruncateEntries(number int) error {
 	return nil
 }
 
-func (idx *Index) InitializePosition() (*Entry, error) {
+func (idx *index) InitializePosition() (*entry, error) {
 	// Find the first empty entry.
 	n := int(idx.size / entryWidth)
-	entry := new(Entry)
+	entry := new(entry)
 	i := sort.Search(n, func(i int) bool {
 		if err := idx.ReadEntryAtFileOffset(entry, int64(i*entryWidth)); err != nil {
 			panic(err)
@@ -271,27 +269,27 @@ func (idx *Index) InitializePosition() (*Entry, error) {
 	}
 	// Do some sanity checks.
 	if entry.Offset < idx.baseOffset {
-		return nil, ErrIndexCorrupt
+		return nil, errIndexCorrupt
 	}
 	idx.mu.RLock()
 	defer idx.mu.RUnlock()
 	if idx.position%entryWidth != 0 {
-		return nil, ErrIndexCorrupt
+		return nil, errIndexCorrupt
 	}
 	return entry, nil
 }
 
-type IndexScanner struct {
-	idx    *Index
-	entry  *Entry
+type indexScanner struct {
+	idx    *index
+	entry  *entry
 	offset int64
 }
 
-func NewIndexScanner(idx *Index) *IndexScanner {
-	return &IndexScanner{idx: idx, entry: &Entry{}}
+func newIndexScanner(idx *index) *indexScanner {
+	return &indexScanner{idx: idx, entry: &entry{}}
 }
 
-func (s *IndexScanner) Scan() (*Entry, error) {
+func (s *indexScanner) Scan() (*entry, error) {
 	err := s.idx.ReadEntryAtLogOffset(s.entry, s.offset)
 	if err != nil {
 		return nil, err
diff --git a/server/commitlog.go b/server/commitlog/interface.go
similarity index 87%
rename from server/commitlog.go
rename to server/commitlog/interface.go
index daeb9e7..76e2d3e 100644
--- a/server/commitlog.go
+++ b/server/commitlog/interface.go
@@ -1,9 +1,6 @@
-package server
+package commitlog
 
-import (
-	"github.com/liftbridge-io/liftbridge/server/commitlog"
-	"github.com/liftbridge-io/liftbridge/server/proto"
-)
+import "github.com/liftbridge-io/liftbridge/server/proto"
 
 // CommitLog is the durable write-ahead log interface used to back each stream.
 type CommitLog interface {
@@ -14,7 +11,7 @@ type CommitLog interface {
 	// NewReader creates a new Reader starting at the given offset. If
 	// uncommitted is true, the Reader will read uncommitted messages from the
 	// log. Otherwise, it will only return committed messages.
-	NewReader(offset int64, uncommitted bool) (*commitlog.Reader, error)
+	NewReader(offset int64, uncommitted bool) (*Reader, error)
 
 	// Truncate removes all messages from the log starting at the given offset.
 	Truncate(offset int64) error
@@ -35,6 +32,11 @@ type CommitLog interface {
 	// and including the high watermark are considered committed.
 	SetHighWatermark(hw int64)
 
+	// OverrideHighWatermark sets the high watermark on the log using the given
+	// value, even if the value is less than the current HW. This is used for
+	// unit testing purposes.
+	OverrideHighWatermark(hw int64)
+
 	// HighWatermark returns the high watermark for the log.
 	HighWatermark() int64
 
diff --git a/server/commitlog/message.go b/server/commitlog/message.go
index 1a3814f..6579560 100644
--- a/server/commitlog/message.go
+++ b/server/commitlog/message.go
@@ -2,20 +2,25 @@ package commitlog
 
 import "github.com/liftbridge-io/liftbridge/server/proto"
 
+// Message is a message read from the log.
 type Message []byte
 
+// Crc returns the CRC32 digest of the message.
 func (m Message) Crc() uint32 {
 	return proto.Encoding.Uint32(m)
 }
 
+// MagicByte returns the byte used for encoding protocol version detection.
 func (m Message) MagicByte() int8 {
 	return int8(m[4])
 }
 
+// Attributes returns the byte used for message flags.
 func (m Message) Attributes() int8 {
 	return int8(m[5])
 }
 
+// Key returns the message key.
 func (m Message) Key() []byte {
 	start, end, size := m.keyOffsets()
 	if size == -1 {
@@ -24,6 +29,7 @@ func (m Message) Key() []byte {
 	return m[start+4 : end]
 }
 
+// Value returns the message value.
 func (m Message) Value() []byte {
 	start, end, size := m.valueOffsets()
 	if size == -1 {
@@ -32,6 +38,7 @@ func (m Message) Value() []byte {
 	return m[start+4 : end]
 }
 
+// Headers returns the message headers map.
 func (m Message) Headers() map[string][]byte {
 	var (
 		_, valueEnd, _ = m.valueOffsets()
diff --git a/server/commitlog/message_set.go b/server/commitlog/message_set.go
index 232ad7a..be7bdb2 100644
--- a/server/commitlog/message_set.go
+++ b/server/commitlog/message_set.go
@@ -20,10 +20,10 @@ const (
 	msgSetHeaderLen = 28
 )
 
-type MessageSet []byte
+type messageSet []byte
 
-func EntriesForMessageSet(basePos int64, ms []byte) []*Entry {
-	entries := []*Entry{}
+func entriesForMessageSet(basePos int64, ms []byte) []*entry {
+	entries := []*entry{}
 	if len(ms) <= msgSetHeaderLen {
 		return entries
 	}
@@ -31,13 +31,13 @@ func EntriesForMessageSet(basePos int64, ms []byte) []*Entry {
 	for len(ms) > 0 {
 		var (
 			relPos      = n
-			m           = MessageSet(ms)
+			m           = messageSet(ms)
 			offset      = m.Offset()
 			timestamp   = m.Timestamp()
 			leaderEpoch = m.LeaderEpoch()
 			size        = m.Size()
 		)
-		entries = append(entries, &Entry{
+		entries = append(entries, &entry{
 			Offset:      offset,
 			Timestamp:   timestamp,
 			LeaderEpoch: leaderEpoch,
@@ -50,12 +50,12 @@ func EntriesForMessageSet(basePos int64, ms []byte) []*Entry {
 	return entries
 }
 
-func NewMessageSetFromProto(baseOffset, basePos int64, msgs []*proto.Message) (
-	MessageSet, []*Entry, error) {
+func newMessageSetFromProto(baseOffset, basePos int64, msgs []*proto.Message) (
+	messageSet, []*entry, error) {
 
 	var (
 		buf     = new(bytes.Buffer)
-		entries = make([]*Entry, len(msgs))
+		entries = make([]*entry, len(msgs))
 		n       int32
 	)
 	for i, m := range msgs {
@@ -88,7 +88,7 @@ func NewMessageSetFromProto(baseOffset, basePos int64, msgs []*proto.Message) (
 			return nil, nil, err
 		}
 		n += len
-		entries[i] = &Entry{
+		entries[i] = &entry{
 			Offset:      offset,
 			Timestamp:   m.Timestamp,
 			LeaderEpoch: m.LeaderEpoch,
@@ -128,23 +128,23 @@ func readMessage(ctx context.Context, reader contextReader, headersBuf []byte) (
 	return m, offset, timestamp, leaderEpoch, nil
 }
 
-func (ms MessageSet) Offset() int64 {
+func (ms messageSet) Offset() int64 {
 	return int64(proto.Encoding.Uint64(ms[offsetPos : offsetPos+8]))
 }
 
-func (ms MessageSet) Timestamp() int64 {
+func (ms messageSet) Timestamp() int64 {
 	return int64(proto.Encoding.Uint64(ms[timestampPos : timestampPos+8]))
 }
 
-func (ms MessageSet) LeaderEpoch() uint64 {
+func (ms messageSet) LeaderEpoch() uint64 {
 	return proto.Encoding.Uint64(ms[leaderEpochPos : leaderEpochPos+8])
 }
 
-func (ms MessageSet) Size() int32 {
+func (ms messageSet) Size() int32 {
 	return int32(proto.Encoding.Uint32(ms[sizePos : sizePos+4]))
 }
 
-func (ms MessageSet) Message() Message {
+func (ms messageSet) Message() Message {
 	if len(ms) <= msgSetHeaderLen {
 		return nil
 	}
diff --git a/server/commitlog/reader.go b/server/commitlog/reader.go
index 58cc3a9..df17467 100644
--- a/server/commitlog/reader.go
+++ b/server/commitlog/reader.go
@@ -18,14 +18,14 @@ type contextReader interface {
 type Reader struct {
 	ctxReader   contextReader
 	offset      int64
-	log         *CommitLog
+	log         *commitLog
 	uncommitted bool
 }
 
 // NewReader creates a new Reader starting at the given offset. If uncommitted
 // is true, the Reader will read uncommitted messages from the log. Otherwise,
 // it will only return committed messages.
-func (l *CommitLog) NewReader(offset int64, uncommitted bool) (*Reader, error) {
+func (l *commitLog) NewReader(offset int64, uncommitted bool) (*Reader, error) {
 	var (
 		ctxReader contextReader
 		err       error
@@ -79,8 +79,8 @@ RETRY:
 }
 
 type uncommittedReader struct {
-	cl  *CommitLog
-	seg *Segment
+	cl  *commitLog
+	seg *segment
 	mu  sync.Mutex
 	pos int64
 }
@@ -155,7 +155,7 @@ LOOP:
 	return n, err
 }
 
-func (r *uncommittedReader) waitForData(ctx context.Context, seg *Segment) bool {
+func (r *uncommittedReader) waitForData(ctx context.Context, seg *segment) bool {
 	wait := seg.WaitForData(r, r.pos)
 	select {
 	case <-r.cl.closed:
@@ -171,7 +171,7 @@ func (r *uncommittedReader) waitForData(ctx context.Context, seg *Segment) bool
 
 // newReaderUncommitted returns a contextReader which reads data from the log
 // starting at the given offset.
-func (l *CommitLog) newReaderUncommitted(offset int64) (contextReader, error) {
+func (l *commitLog) newReaderUncommitted(offset int64) (contextReader, error) {
 	seg, contains := findSegmentContains(l.Segments(), offset)
 	if seg == nil {
 		return nil, ErrSegmentNotFound
@@ -192,9 +192,9 @@ func (l *CommitLog) newReaderUncommitted(offset int64) (contextReader, error) {
 }
 
 type committedReader struct {
-	cl    *CommitLog
-	seg   *Segment
-	hwSeg *Segment
+	cl    *commitLog
+	seg   *segment
+	hwSeg *segment
 	mu    sync.Mutex
 	pos   int64
 	hwPos int64
@@ -244,7 +244,7 @@ func (r *committedReader) Read(ctx context.Context, p []byte) (n int, err error)
 }
 
 func (r *committedReader) readLoop(
-	ctx context.Context, p []byte, segments []*Segment) (n int, err error) {
+	ctx context.Context, p []byte, segments []*segment) (n int, err error) {
 
 	var readSize int
 LOOP:
@@ -320,12 +320,12 @@ func (r *committedReader) waitForHW(ctx context.Context, hw int64) bool {
 
 // newReaderCommitted returns a contextReader which reads only committed data
 // from the log starting at the given offset.
-func (l *CommitLog) newReaderCommitted(offset int64) (contextReader, error) {
+func (l *commitLog) newReaderCommitted(offset int64) (contextReader, error) {
 	var (
 		hw       = l.HighWatermark()
 		hwPos    = int64(-1)
 		segments = l.Segments()
-		hwSeg    *Segment
+		hwSeg    *segment
 	)
 
 	// If offset exceeds HW, wait for the next message. This also covers the
@@ -369,7 +369,7 @@ func (l *CommitLog) newReaderCommitted(offset int64) (contextReader, error) {
 	}, nil
 }
 
-func getHWPos(segments []*Segment, hw int64) (int, int64, error) {
+func getHWPos(segments []*segment, hw int64) (int, int64, error) {
 	hwSeg, hwIdx := findSegment(segments, hw)
 	if hwSeg == nil {
 		return 0, 0, ErrSegmentNotFound
diff --git a/server/commitlog/segment.go b/server/commitlog/segment.go
index bdd06a8..1aee847 100644
--- a/server/commitlog/segment.go
+++ b/server/commitlog/segment.go
@@ -43,11 +43,11 @@ var (
 	timestamp = func() int64 { return time.Now().UnixNano() }
 )
 
-type Segment struct {
+type segment struct {
 	writer         io.Writer
 	reader         io.Reader
 	log            *os.File
-	Index          *Index
+	Index          *index
 	BaseOffset     int64
 	firstOffset    int64
 	lastOffset     int64
@@ -65,8 +65,8 @@ type Segment struct {
 	sync.RWMutex
 }
 
-func NewSegment(path string, baseOffset, maxBytes int64, isNew bool, suffix string) (*Segment, error) {
-	s := &Segment{
+func newSegment(path string, baseOffset, maxBytes int64, isNew bool, suffix string) (*segment, error) {
+	s := &segment{
 		maxBytes:    maxBytes,
 		BaseOffset:  baseOffset,
 		firstOffset: -1,
@@ -95,13 +95,13 @@ func NewSegment(path string, baseOffset, maxBytes int64, isNew bool, suffix stri
 	return s, err
 }
 
-// setupIndex creates and initializes an Index.
+// setupIndex creates and initializes an index.
 // Initialization is:
-// - Initialize Index position
+// - Initialize index position
 // - Initialize firstOffset/lastOffset
 // - Initialize firstWriteTime/lastWriteTime
-func (s *Segment) setupIndex() (err error) {
-	s.Index, err = NewIndex(options{
+func (s *segment) setupIndex() (err error) {
+	s.Index, err = newIndex(options{
 		path:       s.indexPath(),
 		baseOffset: s.BaseOffset,
 	})
@@ -117,7 +117,7 @@ func (s *Segment) setupIndex() (err error) {
 		s.lastOffset = lastEntry.Offset
 		s.lastWriteTime = lastEntry.Timestamp
 		// Read the first entry to get firstOffset and firstWriteTime.
-		var firstEntry Entry
+		var firstEntry entry
 		if err := s.Index.ReadEntryAtFileOffset(&firstEntry, 0); err != nil {
 			return err
 		}
@@ -130,7 +130,7 @@ func (s *Segment) setupIndex() (err error) {
 // CheckSplit determines if a new log segment should be rolled out either
 // because this segment is full or LogRollTime has passed since the first
 // message was written to the segment.
-func (s *Segment) CheckSplit(logRollTime time.Duration) bool {
+func (s *segment) CheckSplit(logRollTime time.Duration) bool {
 	s.RLock()
 	defer s.RUnlock()
 	if s.position >= s.maxBytes {
@@ -148,7 +148,7 @@ func (s *Segment) CheckSplit(logRollTime time.Duration) bool {
 // Seal a segment from being written to. This is called on the former active
 // segment after a new segment is rolled. This is a no-op if the segment is
 // already sealed.
-func (s *Segment) Seal() {
+func (s *segment) Seal() {
 	s.Lock()
 	defer s.Unlock()
 	if s.sealed {
@@ -160,7 +160,7 @@ func (s *Segment) Seal() {
 	s.Index.Shrink() // nolint: errcheck
 }
 
-func (s *Segment) NextOffset() int64 {
+func (s *segment) NextOffset() int64 {
 	s.RLock()
 	defer s.RUnlock()
 	// If the segment hasn't been written to, the next offset should be the
@@ -171,37 +171,37 @@ func (s *Segment) NextOffset() int64 {
 	return s.lastOffset + 1
 }
 
-func (s *Segment) FirstOffset() int64 {
+func (s *segment) FirstOffset() int64 {
 	s.RLock()
 	defer s.RUnlock()
 	return s.firstOffset
 }
 
-func (s *Segment) LastOffset() int64 {
+func (s *segment) LastOffset() int64 {
 	s.RLock()
 	defer s.RUnlock()
 	return s.lastOffset
 }
 
-func (s *Segment) Position() int64 {
+func (s *segment) Position() int64 {
 	s.RLock()
 	defer s.RUnlock()
 	return s.position
 }
 
-func (s *Segment) IsEmpty() bool {
+func (s *segment) IsEmpty() bool {
 	s.RLock()
 	defer s.RUnlock()
 	return s.firstOffset == -1
 }
 
-func (s *Segment) MessageCount() int64 {
+func (s *segment) MessageCount() int64 {
 	s.RLock()
 	defer s.RUnlock()
 	return s.Index.CountEntries()
 }
 
-func (s *Segment) WriteMessageSet(ms []byte, entries []*Entry) error {
+func (s *segment) WriteMessageSet(ms []byte, entries []*entry) error {
 	s.Lock()
 	defer s.Unlock()
 	if _, err := s.write(ms, entries); err != nil {
@@ -212,7 +212,7 @@ func (s *Segment) WriteMessageSet(ms []byte, entries []*Entry) error {
 
 // write a byte slice to the log at the current position. This increments the
 // offset as well as sets the position to the new tail.
-func (s *Segment) write(p []byte, entries []*Entry) (n int, err error) {
+func (s *segment) write(p []byte, entries []*entry) (n int, err error) {
 	if s.closed {
 		return 0, ErrSegmentClosed
 	}
@@ -233,7 +233,7 @@ func (s *Segment) write(p []byte, entries []*Entry) (n int, err error) {
 	return n, nil
 }
 
-func (s *Segment) ReadAt(p []byte, off int64) (n int, err error) {
+func (s *segment) ReadAt(p []byte, off int64) (n int, err error) {
 	s.RLock()
 	defer s.RUnlock()
 	if s.closed {
@@ -245,14 +245,14 @@ func (s *Segment) ReadAt(p []byte, off int64) (n int, err error) {
 	return s.log.ReadAt(p, off)
 }
 
-func (s *Segment) notifyWaiters() {
+func (s *segment) notifyWaiters() {
 	for r, ch := range s.waiters {
 		close(ch)
 		delete(s.waiters, r)
 	}
 }
 
-func (s *Segment) WaitForLEO(waiter interface{}, leo int64) <-chan struct{} {
+func (s *segment) WaitForLEO(waiter interface{}, leo int64) <-chan struct{} {
 	s.Lock()
 	defer s.Unlock()
 	if s.lastOffset != leo {
@@ -262,14 +262,14 @@ func (s *Segment) WaitForLEO(waiter interface{}, leo int64) <-chan struct{} {
 	}
 	return s.waitForData(waiter, s.position)
 }
-func (s *Segment) WaitForData(waiter interface{}, pos int64) <-chan struct{} {
+func (s *segment) WaitForData(waiter interface{}, pos int64) <-chan struct{} {
 	s.Lock()
 	ch := s.waitForData(waiter, pos)
 	s.Unlock()
 	return ch
 }
 
-func (s *Segment) waitForData(waiter interface{}, pos int64) <-chan struct{} {
+func (s *segment) waitForData(waiter interface{}, pos int64) <-chan struct{} {
 	// Check if we're already registered.
 	wait, ok := s.waiters[waiter]
 	if ok {
@@ -285,7 +285,7 @@ func (s *Segment) waitForData(waiter interface{}, pos int64) <-chan struct{} {
 	return wait
 }
 
-func (s *Segment) removeWaiter(waiter interface{}) {
+func (s *segment) removeWaiter(waiter interface{}) {
 	s.Lock()
 	delete(s.waiters, waiter)
 	s.Unlock()
@@ -293,13 +293,13 @@ func (s *Segment) removeWaiter(waiter interface{}) {
 
 // Close a segment such that it can no longer be read from or written to. This
 // operation is idempotent.
-func (s *Segment) Close() error {
+func (s *segment) Close() error {
 	s.Lock()
 	defer s.Unlock()
 	return s.close()
 }
 
-func (s *Segment) close() error {
+func (s *segment) close() error {
 	if s.closed {
 		return nil
 	}
@@ -314,17 +314,17 @@ func (s *Segment) close() error {
 }
 
 // Cleaned creates a cleaned segment for this segment.
-func (s *Segment) Cleaned() (*Segment, error) {
-	return NewSegment(s.path, s.BaseOffset, s.maxBytes, false, cleanedSuffix)
+func (s *segment) Cleaned() (*segment, error) {
+	return newSegment(s.path, s.BaseOffset, s.maxBytes, false, cleanedSuffix)
 }
 
 // Truncated creates a truncated segment for this segment.
-func (s *Segment) Truncated() (*Segment, error) {
-	return NewSegment(s.path, s.BaseOffset, s.maxBytes, false, truncatedSuffix)
+func (s *segment) Truncated() (*segment, error) {
+	return newSegment(s.path, s.BaseOffset, s.maxBytes, false, truncatedSuffix)
 }
 
 // Replace replaces the given segment with the callee.
-func (s *Segment) Replace(old *Segment) error {
+func (s *segment) Replace(old *segment) error {
 	s.Lock()
 	defer s.Unlock()
 	old.Lock()
@@ -356,10 +356,10 @@ func (s *Segment) Replace(old *Segment) error {
 
 // findEntry returns the first entry whose offset is greater than or equal to
 // the given offset.
-func (s *Segment) findEntry(offset int64) (e *Entry, err error) {
+func (s *segment) findEntry(offset int64) (e *entry, err error) {
 	s.RLock()
 	defer s.RUnlock()
-	e = &Entry{}
+	e = &entry{}
 	n := int(s.Index.Position() / entryWidth)
 	idx := sort.Search(n, func(i int) bool {
 		if err := s.Index.ReadEntryAtFileOffset(e, int64(i*entryWidth)); err != nil {
@@ -376,10 +376,10 @@ func (s *Segment) findEntry(offset int64) (e *Entry, err error) {
 
 // findEntryByTimestamp returns the first entry whose timestamp is greater than
 // or equal to the given offset.
-func (s *Segment) findEntryByTimestamp(timestamp int64) (e *Entry, err error) {
+func (s *segment) findEntryByTimestamp(timestamp int64) (e *entry, err error) {
 	s.RLock()
 	defer s.RUnlock()
-	e = &Entry{}
+	e = &entry{}
 	n := int(s.Index.Position() / entryWidth)
 	idx := sort.Search(n, func(i int) bool {
 		if err := s.Index.ReadEntryAtFileOffset(e, int64(i*entryWidth)); err != nil {
@@ -395,7 +395,7 @@ func (s *Segment) findEntryByTimestamp(timestamp int64) (e *Entry, err error) {
 }
 
 // Delete closes the segment and then deletes its log and index files.
-func (s *Segment) Delete() error {
+func (s *segment) Delete() error {
 	if err := s.Close(); err != nil {
 		return err
 	}
@@ -414,23 +414,23 @@ func (s *Segment) Delete() error {
 	return nil
 }
 
-type SegmentScanner struct {
-	s  *Segment
-	is *IndexScanner
+type segmentScanner struct {
+	s  *segment
+	is *indexScanner
 }
 
-func NewSegmentScanner(segment *Segment) *SegmentScanner {
-	return &SegmentScanner{s: segment, is: NewIndexScanner(segment.Index)}
+func newSegmentScanner(segment *segment) *segmentScanner {
+	return &segmentScanner{s: segment, is: newIndexScanner(segment.Index)}
 }
 
 // Scan should be called repeatedly to iterate over the messages in the
 // segment, it will return io.EOF when there are no more messages.
-func (s *SegmentScanner) Scan() (MessageSet, *Entry, error) {
+func (s *segmentScanner) Scan() (messageSet, *entry, error) {
 	entry, err := s.is.Scan()
 	if err != nil {
 		return nil, nil, err
 	}
-	header := make(MessageSet, msgSetHeaderLen)
+	header := make(messageSet, msgSetHeaderLen)
 	_, err = s.s.ReadAt(header, entry.Position)
 	if err != nil {
 		return nil, nil, err
@@ -444,10 +444,10 @@ func (s *SegmentScanner) Scan() (MessageSet, *Entry, error) {
 	return msgSet, entry, nil
 }
 
-func (s *Segment) logPath() string {
+func (s *segment) logPath() string {
 	return filepath.Join(s.path, fmt.Sprintf(fileFormat, s.BaseOffset, logSuffix+s.suffix))
 }
 
-func (s *Segment) indexPath() string {
+func (s *segment) indexPath() string {
 	return filepath.Join(s.path, fmt.Sprintf(fileFormat, s.BaseOffset, indexSuffix+s.suffix))
 }
diff --git a/server/commitlog/util.go b/server/commitlog/util.go
index 3bc7102..e9ae0e8 100644
--- a/server/commitlog/util.go
+++ b/server/commitlog/util.go
@@ -8,7 +8,7 @@ import (
 // findSegment returns the first segment whose next assignable offset is
 // greater than the given offset. Returns nil and the index where the segment
 // would be if there is no such segment.
-func findSegment(segments []*Segment, offset int64) (*Segment, int) {
+func findSegment(segments []*segment, offset int64) (*segment, int) {
 	n := len(segments)
 	idx := sort.Search(n, func(i int) bool {
 		return segments[i].NextOffset() > offset
@@ -25,7 +25,7 @@ func findSegment(segments []*Segment, offset int64) (*Segment, int) {
 // base offset and next assignable offset. Note that because the segment could
 // be compacted, "contains" does not guarantee the offset is actually present,
 // only that it's within the bounds.
-func findSegmentContains(segments []*Segment, offset int64) (*Segment, bool) {
+func findSegmentContains(segments []*segment, offset int64) (*segment, bool) {
 	seg, _ := findSegment(segments, offset)
 	if seg == nil {
 		return nil, false
@@ -37,14 +37,14 @@ func findSegmentContains(segments []*Segment, offset int64) (*Segment, bool) {
 // base timestamp is greater than the given timestamp. Returns the index where
 // the segment would be if there is no segment whose base timestamp is greater,
 // i.e. the length of the slice.
-func findSegmentIndexByTimestamp(segments []*Segment, timestamp int64) (int, error) {
+func findSegmentIndexByTimestamp(segments []*segment, timestamp int64) (int, error) {
 	var (
 		n   = len(segments)
 		err error
 	)
 	idx := sort.Search(n, func(i int) bool {
 		// Read the first entry in the segment to determine the base timestamp.
-		var entry Entry
+		var entry entry
 		if e := segments[i].Index.ReadEntryAtLogOffset(&entry, 0); e != nil {
 			err = e
 			return true
@@ -57,7 +57,7 @@ func findSegmentIndexByTimestamp(segments []*Segment, timestamp int64) (int, err
 // findSegmentByBaseOffset returns the first segment whose base offset is
 // greater than or equal to the given offset. Returns nil if there is no such
 // segment.
-func findSegmentByBaseOffset(segments []*Segment, offset int64) *Segment {
+func findSegmentByBaseOffset(segments []*segment, offset int64) *segment {
 	n := len(segments)
 	idx := sort.Search(n, func(i int) bool {
 		return segments[i].BaseOffset >= offset
diff --git a/server/conf/lex.go b/server/conf/lex.go
index 4eab7d2..7dc6856 100644
--- a/server/conf/lex.go
+++ b/server/conf/lex.go
@@ -1088,6 +1088,7 @@ func isNL(r rune) bool {
 	return r == '\n' || r == '\r'
 }
 
+// nolint: gocyclo
 func (itype itemType) String() string {
 	switch itype {
 	case itemError:
diff --git a/server/conf/parse.go b/server/conf/parse.go
index 0166b50..763a22c 100644
--- a/server/conf/parse.go
+++ b/server/conf/parse.go
@@ -135,6 +135,7 @@ func (p *parser) popKey() string {
 	return last
 }
 
+// nolint: gocyclo
 func (p *parser) processItem(it item) error {
 	switch it.typ {
 	case itemError:
diff --git a/server/config.go b/server/config.go
index faf048f..bc7c751 100644
--- a/server/config.go
+++ b/server/config.go
@@ -202,7 +202,7 @@ func GetLogLevel(level string) (uint32, error) {
 
 // NewConfig creates a new Config with default settings and applies any
 // settings from the given configuration file.
-func NewConfig(configFile string) (*Config, error) {
+func NewConfig(configFile string) (*Config, error) { // nolint: gocyclo
 	config := NewDefaultConfig()
 
 	if configFile == "" {
@@ -352,7 +352,7 @@ func parseLogConfig(config *Config, m map[string]interface{}) error {
 
 // parseClusteringConfig parses the `clustering` section of a config file and
 // populates the given Config.
-func parseClusteringConfig(config *Config, m map[string]interface{}) error {
+func parseClusteringConfig(config *Config, m map[string]interface{}) error { // nolint: gocyclo
 	for k, v := range m {
 		switch strings.ToLower(k) {
 		case "server.id":
diff --git a/server/metadata.go b/server/metadata.go
index cb17e0f..d29ff17 100644
--- a/server/metadata.go
+++ b/server/metadata.go
@@ -174,7 +174,7 @@ func (m *metadataAPI) brokerCache(serverIDs map[string]struct{}) ([]*client.Brok
 func (m *metadataAPI) fetchBrokerInfo(ctx context.Context, numPeers int) ([]*client.Broker, *status.Status) {
 	// Add ourselves.
 	connectionAddress := m.config.GetConnectionAddress()
-	brokers := []*client.Broker{&client.Broker{
+	brokers := []*client.Broker{{
 		Id:   m.config.Clustering.ServerID,
 		Host: connectionAddress.Host,
 		Port: int32(connectionAddress.Port),
diff --git a/server/partition.go b/server/partition.go
index fddd27b..6538190 100644
--- a/server/partition.go
+++ b/server/partition.go
@@ -76,7 +76,7 @@ type partition struct {
 	leaderReplSub   *nats.Subscription // Subscription for replication requests from followers
 	leaderOffsetSub *nats.Subscription // Subscription for leader epoch offset requests from followers
 	recvChan        chan *nats.Msg     // Channel leader places received messages on
-	log             CommitLog
+	log             commitlog.CommitLog
 	srv             *Server
 	isLeading       bool
 	isFollowing     bool
diff --git a/server/proto/crc_field.go b/server/proto/crc_field.go
index cc249c2..22282a6 100644
--- a/server/proto/crc_field.go
+++ b/server/proto/crc_field.go
@@ -5,24 +5,29 @@ import (
 	"hash/crc32"
 )
 
+// CRCField is used to perform a CRC32 check on a message.
 type CRCField struct {
 	StartOffset int
 }
 
+// SaveOffset sets the position to fill the CRC digest.
 func (f *CRCField) SaveOffset(in int) {
 	f.StartOffset = in
 }
 
+// ReserveSize sets the number of bytes to reserve for the CRC digest.
 func (f *CRCField) ReserveSize() int {
 	return 4
 }
 
+// Fill sets the CRC digest.
 func (f *CRCField) Fill(curOffset int, buf []byte) error {
 	crc := crc32.ChecksumIEEE(buf[f.StartOffset+4 : curOffset])
 	Encoding.PutUint32(buf[f.StartOffset:], crc)
 	return nil
 }
 
+// Check the CRC digest.
 func (f *CRCField) Check(curOffset int, buf []byte) error {
 	crc := crc32.ChecksumIEEE(buf[f.StartOffset+4 : curOffset])
 	if crc != Encoding.Uint32(buf[f.StartOffset:]) {
diff --git a/server/proto/decoder.go b/server/proto/decoder.go
deleted file mode 100644
index 51d6781..0000000
--- a/server/proto/decoder.go
+++ /dev/null
@@ -1,308 +0,0 @@
-package proto
-
-import (
-	"errors"
-	"math"
-)
-
-var ErrInsufficientData = errors.New("insufficient data to decode packet, more bytes expected")
-var ErrInvalidStringLength = errors.New("invalid string length")
-var ErrInvalidArrayLength = errors.New("invalid array length")
-var ErrInvalidByteSliceLength = errors.New("invalid byteslice length")
-
-type PacketDecoder interface {
-	Bool() (bool, error)
-	Int8() (int8, error)
-	Int16() (int16, error)
-	Int32() (int32, error)
-	Int64() (int64, error)
-	ArrayLength() (int, error)
-	Bytes() ([]byte, error)
-	String() (string, error)
-	NullableString() (*string, error)
-	Int32Array() ([]int32, error)
-	Int64Array() ([]int64, error)
-	StringArray() ([]string, error)
-	Push(pd PushDecoder) error
-	Pop() error
-	remaining() int
-}
-
-type Decoder interface {
-	Decode(d PacketDecoder) error
-}
-
-type VersionedDecoder interface {
-	Decode(d PacketDecoder, version int16) error
-}
-
-type PushDecoder interface {
-	SaveOffset(in int)
-	ReserveSize() int
-	Fill(curOffset int, buf []byte) error
-	Check(curOffset int, buf []byte) error
-}
-
-func Decode(b []byte, in VersionedDecoder, version int16) error {
-	d := NewDecoder(b)
-	return in.Decode(d, version)
-}
-
-type ByteDecoder struct {
-	b     []byte
-	off   int
-	stack []PushDecoder
-}
-
-func (d *ByteDecoder) Offset() int {
-	return d.off
-}
-
-func NewDecoder(b []byte) *ByteDecoder {
-	return &ByteDecoder{
-		b: b,
-	}
-}
-
-func (d *ByteDecoder) Bool() (bool, error) {
-	i, err := d.Int8()
-	if err != nil {
-		return false, err
-	}
-	return i == 1, nil
-}
-
-func (d *ByteDecoder) Int8() (int8, error) {
-	tmp := int8(d.b[d.off])
-	d.off++
-	return tmp, nil
-}
-
-func (d *ByteDecoder) Int16() (int16, error) {
-	tmp := int16(Encoding.Uint16(d.b[d.off:]))
-	d.off += 2
-	return tmp, nil
-}
-
-func (d *ByteDecoder) Int32() (int32, error) {
-	if d.remaining() < 4 {
-		d.off = len(d.b)
-		return -1, ErrInsufficientData
-	}
-	tmp := int32(Encoding.Uint32(d.b[d.off:]))
-	d.off += 4
-	return tmp, nil
-}
-
-func (d *ByteDecoder) Int64() (int64, error) {
-	if d.remaining() < 8 {
-		d.off = len(d.b)
-		return -1, ErrInsufficientData
-	}
-	tmp := int64(Encoding.Uint64(d.b[d.off:]))
-	d.off += 8
-	return tmp, nil
-}
-
-func (d *ByteDecoder) ArrayLength() (int, error) {
-	if d.remaining() < 4 {
-		d.off = len(d.b)
-		return -1, ErrInsufficientData
-	}
-	tmp := int(Encoding.Uint32(d.b[d.off:]))
-	d.off += 4
-	if tmp > d.remaining() {
-		d.off = len(d.b)
-		return -1, ErrInsufficientData
-	} else if tmp > 2*math.MaxUint16 {
-		return -1, ErrInvalidArrayLength
-	}
-	return tmp, nil
-}
-
-// collections
-
-func (d *ByteDecoder) Bytes() ([]byte, error) {
-	tmp, err := d.Int32()
-
-	if err != nil {
-		return nil, err
-	}
-
-	n := int(tmp)
-
-	switch {
-	case n < -1:
-		return nil, ErrInvalidByteSliceLength
-	case n == -1:
-		return nil, nil
-	case n == 0:
-		return make([]byte, 0), nil
-	case n > d.remaining():
-		d.off = len(d.b)
-		return nil, ErrInsufficientData
-	}
-
-	tmpStr := d.b[d.off : d.off+n]
-	d.off += n
-	return tmpStr, nil
-}
-
-func (d *ByteDecoder) String() (string, error) {
-	tmp, err := d.Int16()
-
-	if err != nil {
-		return "", err
-	}
-
-	n := int(tmp)
-
-	switch {
-	case n < -1:
-		return "", ErrInvalidStringLength
-	case n == -1:
-		return "", nil
-	case n == 0:
-		return "", nil
-	case n > d.remaining():
-		d.off = len(d.b)
-		return "", ErrInsufficientData
-	}
-
-	tmpStr := string(d.b[d.off : d.off+n])
-	d.off += n
-	return tmpStr, nil
-}
-
-func (d *ByteDecoder) stringLength() (int, error) {
-	l, err := d.Int16()
-	if err != nil {
-		return 0, err
-	}
-	n := int(l)
-	switch {
-	case n < -1:
-		return 0, ErrInvalidStringLength
-	case n > d.remaining():
-		d.off = len(d.b)
-		return 0, ErrInsufficientData
-	}
-	return n, nil
-}
-
-func (d *ByteDecoder) NullableString() (*string, error) {
-	n, err := d.stringLength()
-	if err != nil || n == -1 {
-		return nil, err
-	}
-	tmpStr := string(d.b[d.off : d.off+n])
-	d.off += n
-	return &tmpStr, nil
-}
-
-func (d *ByteDecoder) Int32Array() ([]int32, error) {
-	if d.remaining() < 4 {
-		d.off = len(d.b)
-		return nil, ErrInsufficientData
-	}
-	n := int(Encoding.Uint32(d.b[d.off:]))
-	d.off += 4
-
-	if d.remaining() < 4*n {
-		d.off = len(d.b)
-		return nil, ErrInsufficientData
-	}
-
-	if n == 0 {
-		return nil, nil
-	}
-
-	if n < 0 {
-		return nil, ErrInvalidArrayLength
-	}
-
-	ret := make([]int32, n)
-	for i := range ret {
-		ret[i] = int32(Encoding.Uint32(d.b[d.off:]))
-		d.off += 4
-	}
-	return ret, nil
-}
-
-func (d *ByteDecoder) Int64Array() ([]int64, error) {
-	if d.remaining() < 4 {
-		d.off = len(d.b)
-		return nil, ErrInsufficientData
-	}
-	n := int(Encoding.Uint32(d.b[d.off:]))
-	d.off += 4
-
-	if d.remaining() < 8*n {
-		d.off = len(d.b)
-		return nil, ErrInsufficientData
-	}
-
-	if n == 0 {
-		return nil, nil
-	}
-
-	if n < 0 {
-		return nil, ErrInvalidArrayLength
-	}
-
-	ret := make([]int64, n)
-	for i := range ret {
-		ret[i] = int64(Encoding.Uint64(d.b[d.off:]))
-		d.off += 8
-	}
-	return ret, nil
-}
-
-func (d *ByteDecoder) StringArray() ([]string, error) {
-	if d.remaining() < 4 {
-		d.off = len(d.b)
-		return nil, ErrInsufficientData
-	}
-	n := int(Encoding.Uint32(d.b[d.off:]))
-	d.off += 4
-
-	if n == 0 {
-		return nil, nil
-	}
-
-	if n < 0 {
-		return nil, ErrInvalidArrayLength
-	}
-
-	ret := make([]string, n)
-	for i := range ret {
-		if str, err := d.String(); err != nil {
-			return nil, err
-		} else {
-			ret[i] = str
-		}
-	}
-	return ret, nil
-}
-
-func (d *ByteDecoder) Push(pd PushDecoder) error {
-	pd.SaveOffset(d.off)
-	reserved := pd.ReserveSize()
-	if d.remaining() < reserved {
-		d.off = len(d.b)
-		return ErrInsufficientData
-	}
-	d.stack = append(d.stack, pd)
-	d.off += reserved
-	return nil
-}
-
-func (d *ByteDecoder) Pop() error {
-	pd := d.stack[len(d.stack)-1]
-	d.stack = d.stack[:len(d.stack)-1]
-	return pd.Check(d.off, d.b)
-}
-
-func (d *ByteDecoder) remaining() int {
-	return len(d.b) - d.off
-}
diff --git a/server/proto/encoder.go b/server/proto/encoder.go
index 51ed7cd..7340a52 100644
--- a/server/proto/encoder.go
+++ b/server/proto/encoder.go
@@ -2,11 +2,20 @@ package proto
 
 import (
 	"encoding/binary"
+	"errors"
 	"math"
 )
 
-var Encoding = binary.BigEndian
+var (
+	// Encoding is the byte order to use for serialization.
+	Encoding = binary.BigEndian
 
+	errInvalidStringLength    = errors.New("invalid string length")
+	errInvalidArrayLength     = errors.New("invalid array length")
+	errInvalidByteSliceLength = errors.New("invalid byteslice length")
+)
+
+// PacketEncoder is used to serialize an object.
 type PacketEncoder interface {
 	PutBool(in bool)
 	PutInt8(in int8)
@@ -25,16 +34,20 @@ type PacketEncoder interface {
 	Pop()
 }
 
+// PushEncoder is used to push an operation onto the stack to perform later
+// once serialized bytes are filled.
 type PushEncoder interface {
 	SaveOffset(in int)
 	ReserveSize() int
 	Fill(curOffset int, buf []byte) error
 }
 
+// Encoder is a struct that can be serialized.
 type Encoder interface {
 	Encode(e PacketEncoder) error
 }
 
+// Encode serializes the struct to bytes.
 func Encode(e Encoder) ([]byte, error) {
 	lenEnc := new(LenEncoder)
 	err := e.Encode(lenEnc)
@@ -52,33 +65,41 @@ func Encode(e Encoder) ([]byte, error) {
 	return b, nil
 }
 
+// LenEncoder is a PacketEncoder that tracks the running length of serialized
+// bytes.
 type LenEncoder struct {
 	Length int
 }
 
+// PutBool increments length for a bool.
 func (e *LenEncoder) PutBool(in bool) {
 	e.Length++
 }
 
+// PutInt8 increments length for an int8.
 func (e *LenEncoder) PutInt8(in int8) {
 	e.Length++
 }
 
+// PutInt16 increments length for an int16.
 func (e *LenEncoder) PutInt16(in int16) {
 	e.Length += 2
 }
 
+// PutInt32 increments length for an int32.
 func (e *LenEncoder) PutInt32(in int32) {
 	e.Length += 4
 }
 
+// PutInt64 increments length for an int64.
 func (e *LenEncoder) PutInt64(in int64) {
 	e.Length += 8
 }
 
+// PutArrayLength increments length for an array size.
 func (e *LenEncoder) PutArrayLength(in int) error {
 	if in > math.MaxInt32 {
-		return ErrInvalidArrayLength
+		return errInvalidArrayLength
 	}
 	e.Length += 4
 	return nil
@@ -86,35 +107,39 @@ func (e *LenEncoder) PutArrayLength(in int) error {
 
 // arrays
 
+// PutBytes increments length for a size-prefixed byte array.
 func (e *LenEncoder) PutBytes(in []byte) error {
 	e.Length += 4
 	if in == nil {
 		return nil
 	}
 	if len(in) > math.MaxInt32 {
-		return ErrInvalidByteSliceLength
+		return errInvalidByteSliceLength
 	}
 	e.Length += len(in)
 	return nil
 }
 
+// PutRawBytes increments length for a raw byte array.
 func (e *LenEncoder) PutRawBytes(in []byte) error {
 	if len(in) > math.MaxInt32 {
-		return ErrInvalidByteSliceLength
+		return errInvalidByteSliceLength
 	}
 	e.Length += len(in)
 	return nil
 }
 
+// PutString increments length for a string.
 func (e *LenEncoder) PutString(in string) error {
 	e.Length += 2
 	if len(in) > math.MaxInt16 {
-		return ErrInvalidStringLength
+		return errInvalidStringLength
 	}
 	e.Length += len(in)
 	return nil
 }
 
+// PutNullableString increments length for a nullable string.
 func (e *LenEncoder) PutNullableString(in *string) error {
 	if in == nil {
 		e.Length += 2
@@ -123,6 +148,7 @@ func (e *LenEncoder) PutNullableString(in *string) error {
 	return e.PutString(*in)
 }
 
+// PutStringArray increments length for a string array.
 func (e *LenEncoder) PutStringArray(in []string) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
@@ -138,6 +164,7 @@ func (e *LenEncoder) PutStringArray(in []string) error {
 	return nil
 }
 
+// PutInt32Array increments length for an int32 array.
 func (e *LenEncoder) PutInt32Array(in []int32) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
@@ -147,6 +174,7 @@ func (e *LenEncoder) PutInt32Array(in []int32) error {
 	return nil
 }
 
+// PutInt64Array increments length for an int64 array.
 func (e *LenEncoder) PutInt64Array(in []int64) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
@@ -156,26 +184,33 @@ func (e *LenEncoder) PutInt64Array(in []int64) error {
 	return nil
 }
 
+// Push increments length based on the PushEncoder's reserved size.
 func (e *LenEncoder) Push(pe PushEncoder) {
 	e.Length += pe.ReserveSize()
 }
 
+// Pop is a no-op.
 func (e *LenEncoder) Pop() {}
 
+// ByteEncoder is a PacketEncoder that serializes data into a byte slice.
 type ByteEncoder struct {
 	b     []byte
 	off   int
 	stack []PushEncoder
 }
 
-func (b *ByteEncoder) Bytes() []byte {
-	return b.b
+// Bytes returns the underlying byte slice.
+func (e *ByteEncoder) Bytes() []byte {
+	return e.b
 }
 
+// NewByteEncoder creates a new ByteEncoder with the given backing
+// pre-allocated byte slice.
 func NewByteEncoder(b []byte) *ByteEncoder {
 	return &ByteEncoder{b: b}
 }
 
+// PutBool serializes a bool.
 func (e *ByteEncoder) PutBool(in bool) {
 	if in {
 		e.b[e.off] = byte(int8(1))
@@ -183,37 +218,44 @@ func (e *ByteEncoder) PutBool(in bool) {
 	e.off++
 }
 
+// PutInt8 serializes an int8.
 func (e *ByteEncoder) PutInt8(in int8) {
 	e.b[e.off] = byte(in)
 	e.off++
 }
 
+// PutInt16 serializes an int16.
 func (e *ByteEncoder) PutInt16(in int16) {
 	Encoding.PutUint16(e.b[e.off:], uint16(in))
 	e.off += 2
 }
 
+// PutInt32 serializes an int32.
 func (e *ByteEncoder) PutInt32(in int32) {
 	Encoding.PutUint32(e.b[e.off:], uint32(in))
 	e.off += 4
 }
 
+// PutInt64 serializes an int64.
 func (e *ByteEncoder) PutInt64(in int64) {
 	Encoding.PutUint64(e.b[e.off:], uint64(in))
 	e.off += 8
 }
 
+// PutArrayLength serializes an array length as an int32.
 func (e *ByteEncoder) PutArrayLength(in int) error {
 	e.PutInt32(int32(in))
 	return nil
 }
 
+// PutRawBytes serializes a byte slice.
 func (e *ByteEncoder) PutRawBytes(in []byte) error {
 	copy(e.b[e.off:], in)
 	e.off += len(in)
 	return nil
 }
 
+// PutBytes serializes a size-prefixed byte slice.
 func (e *ByteEncoder) PutBytes(in []byte) error {
 	if in == nil {
 		e.PutInt32(-1)
@@ -225,6 +267,7 @@ func (e *ByteEncoder) PutBytes(in []byte) error {
 	return nil
 }
 
+// PutString serializes a size-prefixed string.
 func (e *ByteEncoder) PutString(in string) error {
 	e.PutInt16(int16(len(in)))
 	copy(e.b[e.off:], in)
@@ -232,6 +275,7 @@ func (e *ByteEncoder) PutString(in string) error {
 	return nil
 }
 
+// PutNullableString serializes a nullable string.
 func (e *ByteEncoder) PutNullableString(in *string) error {
 	if in == nil {
 		e.PutInt16(-1)
@@ -240,6 +284,7 @@ func (e *ByteEncoder) PutNullableString(in *string) error {
 	return e.PutString(*in)
 }
 
+// PutStringArray serializes a string array.
 func (e *ByteEncoder) PutStringArray(in []string) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
@@ -255,6 +300,7 @@ func (e *ByteEncoder) PutStringArray(in []string) error {
 	return nil
 }
 
+// PutInt32Array serializes an int32 array.
 func (e *ByteEncoder) PutInt32Array(in []int32) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
@@ -266,6 +312,7 @@ func (e *ByteEncoder) PutInt32Array(in []int32) error {
 	return nil
 }
 
+// PutInt64Array serializes an int64 array.
 func (e *ByteEncoder) PutInt64Array(in []int64) error {
 	err := e.PutArrayLength(len(in))
 	if err != nil {
@@ -277,12 +324,15 @@ func (e *ByteEncoder) PutInt64Array(in []int64) error {
 	return nil
 }
 
+// Push adds the given PushEncoder to the stack and saves the current offset
+// position.
 func (e *ByteEncoder) Push(pe PushEncoder) {
 	pe.SaveOffset(e.off)
 	e.off += pe.ReserveSize()
 	e.stack = append(e.stack, pe)
 }
 
+// Pop the stack and run the popped PushEncoder on the serialized data.
 func (e *ByteEncoder) Pop() {
 	// this is go's ugly pop pattern (the inverse of append)
 	pe := e.stack[len(e.stack)-1]
diff --git a/server/proto/message.go b/server/proto/message.go
index 572f0ae..cf4c8f4 100644
--- a/server/proto/message.go
+++ b/server/proto/message.go
@@ -2,6 +2,7 @@ package proto
 
 import client "github.com/liftbridge-io/liftbridge-grpc/go"
 
+// Message is the serialized object that gets written to the log.
 type Message struct {
 	Crc        int32
 	MagicByte  int8
@@ -18,6 +19,7 @@ type Message struct {
 	AckPolicy     client.AckPolicy
 }
 
+// Encode the Message into the PacketEncoder.
 func (m *Message) Encode(e PacketEncoder) error {
 	e.Push(&CRCField{})
 	e.PutInt8(m.MagicByte)
@@ -40,39 +42,3 @@ func (m *Message) Encode(e PacketEncoder) error {
 	e.Pop()
 	return nil
 }
-
-func (m *Message) Decode(d PacketDecoder) error {
-	var err error
-	if err = d.Push(&CRCField{}); err != nil {
-		return err
-	}
-	if m.MagicByte, err = d.Int8(); err != nil {
-		return err
-	}
-	if m.Attributes, err = d.Int8(); err != nil {
-		return err
-	}
-	if m.Key, err = d.Bytes(); err != nil {
-		return err
-	}
-	if m.Value, err = d.Bytes(); err != nil {
-		return err
-	}
-	numHeaders, err := d.Int16()
-	if err != nil {
-		return err
-	}
-	m.Headers = make(map[string][]byte, numHeaders)
-	for i := int16(0); i < numHeaders; i++ {
-		key, err := d.String()
-		if err != nil {
-			return err
-		}
-		value, err := d.Bytes()
-		if err != nil {
-			return err
-		}
-		m.Headers[key] = value
-	}
-	return d.Pop()
-}
diff --git a/server/proto/size_field.go b/server/proto/size_field.go
deleted file mode 100644
index f120b67..0000000
--- a/server/proto/size_field.go
+++ /dev/null
@@ -1,28 +0,0 @@
-package proto
-
-import "errors"
-
-type SizeField struct {
-	StartOffset int
-}
-
-func (s *SizeField) SaveOffset(in int) {
-	s.StartOffset = in
-}
-
-func (s *SizeField) ReserveSize() int {
-	return 4
-}
-
-func (s *SizeField) Fill(curOffset int, buf []byte) error {
-	Encoding.PutUint32(buf[s.StartOffset:], uint32(curOffset-s.StartOffset-4))
-	return nil
-}
-
-func (s *SizeField) Check(curOffset int, buf []byte) error {
-	if uint32(curOffset-s.StartOffset-4) != Encoding.Uint32(buf[s.StartOffset:]) {
-		// replace this error
-		return errors.New("length field invalid")
-	}
-	return nil
-}
diff --git a/server/raft.go b/server/raft.go
index ae9eb0d..277d91e 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -184,7 +184,7 @@ func (s *Server) setupMetadataRaft() error {
 // and with the latter taking precedence.
 func (s *Server) bootstrapCluster(node *raft.Raft) error {
 	// Include ourself in the cluster.
-	servers := []raft.Server{raft.Server{
+	servers := []raft.Server{{
 		ID:      raft.ServerID(s.config.Clustering.ServerID),
 		Address: raft.ServerAddress(s.config.Clustering.ServerID),
 	}}
diff --git a/server/replicator.go b/server/replicator.go
index 7f8af6e..0caf135 100644
--- a/server/replicator.go
+++ b/server/replicator.go
@@ -304,7 +304,7 @@ type replicationProtocolWriter interface {
 type protocolWriter struct {
 	*replicator
 	buf        *bytes.Buffer
-	log        CommitLog
+	log        commitlog.CommitLog
 	lastOffset int64
 	stop       <-chan struct{}
 }
diff --git a/server/server.go b/server/server.go
index 54820c8..5c90597 100644
--- a/server/server.go
+++ b/server/server.go
@@ -515,8 +515,7 @@ func (s *Server) getPropagateInbox() string {
 func (s *Server) handlePropagatedRequest(m *nats.Msg) {
 	var (
 		req  = &proto.PropagatedRequest{}
-		data []byte
-		err  error
+		resp []byte
 	)
 	if err := req.Unmarshal(m.Data); err != nil {
 		s.logger.Warnf("Invalid propagated request: %v", err)
@@ -524,58 +523,78 @@ func (s *Server) handlePropagatedRequest(m *nats.Msg) {
 	}
 	switch req.Op {
 	case proto.Op_CREATE_PARTITION:
-		resp := &proto.PropagatedResponse{
-			Op: req.Op,
-		}
-		if err := s.metadata.CreatePartition(context.Background(), req.CreatePartitionOp); err != nil {
-			resp.Error = &proto.Error{Code: uint32(err.Code()), Msg: err.Message()}
-		}
-		data, err = resp.Marshal()
-		if err != nil {
-			panic(err)
-		}
+		resp = s.handleCreatePartition(req)
 	case proto.Op_SHRINK_ISR:
-		resp := &proto.PropagatedResponse{
-			Op: req.Op,
-		}
-		if err := s.metadata.ShrinkISR(context.Background(), req.ShrinkISROp); err != nil {
-			resp.Error = &proto.Error{Code: uint32(err.Code()), Msg: err.Message()}
-		}
-		data, err = resp.Marshal()
-		if err != nil {
-			panic(err)
-		}
-	case proto.Op_REPORT_LEADER:
-		resp := &proto.PropagatedResponse{
-			Op: req.Op,
-		}
-		if err := s.metadata.ReportLeader(context.Background(), req.ReportLeaderOp); err != nil {
-			resp.Error = &proto.Error{Code: uint32(err.Code()), Msg: err.Message()}
-		}
-		data, err = resp.Marshal()
-		if err != nil {
-			panic(err)
-		}
+		resp = s.handleShrinkISR(req)
 	case proto.Op_EXPAND_ISR:
-		resp := &proto.PropagatedResponse{
-			Op: req.Op,
-		}
-		if err := s.metadata.ExpandISR(context.Background(), req.ExpandISROp); err != nil {
-			resp.Error = &proto.Error{Code: uint32(err.Code()), Msg: err.Message()}
-		}
-		data, err = resp.Marshal()
-		if err != nil {
-			panic(err)
-		}
+		resp = s.handleExpandISR(req)
+	case proto.Op_REPORT_LEADER:
+		resp = s.handleReportLeader(req)
 	default:
 		s.logger.Warnf("Unknown propagated request operation: %s", req.Op)
 		return
 	}
-	if err := m.Respond(data); err != nil {
+	if err := m.Respond(resp); err != nil {
 		s.logger.Errorf("Failed to respond to propagated request: %v", err)
 	}
 }
 
+func (s *Server) handleCreatePartition(req *proto.PropagatedRequest) []byte {
+	resp := &proto.PropagatedResponse{
+		Op: req.Op,
+	}
+	if err := s.metadata.CreatePartition(context.Background(), req.CreatePartitionOp); err != nil {
+		resp.Error = &proto.Error{Code: uint32(err.Code()), Msg: err.Message()}
+	}
+	data, err := resp.Marshal()
+	if err != nil {
+		panic(err)
+	}
+	return data
+}
+
+func (s *Server) handleShrinkISR(req *proto.PropagatedRequest) []byte {
+	resp := &proto.PropagatedResponse{
+		Op: req.Op,
+	}
+	if err := s.metadata.ShrinkISR(context.Background(), req.ShrinkISROp); err != nil {
+		resp.Error = &proto.Error{Code: uint32(err.Code()), Msg: err.Message()}
+	}
+	data, err := resp.Marshal()
+	if err != nil {
+		panic(err)
+	}
+	return data
+}
+
+func (s *Server) handleExpandISR(req *proto.PropagatedRequest) []byte {
+	resp := &proto.PropagatedResponse{
+		Op: req.Op,
+	}
+	if err := s.metadata.ExpandISR(context.Background(), req.ExpandISROp); err != nil {
+		resp.Error = &proto.Error{Code: uint32(err.Code()), Msg: err.Message()}
+	}
+	data, err := resp.Marshal()
+	if err != nil {
+		panic(err)
+	}
+	return data
+}
+
+func (s *Server) handleReportLeader(req *proto.PropagatedRequest) []byte {
+	resp := &proto.PropagatedResponse{
+		Op: req.Op,
+	}
+	if err := s.metadata.ReportLeader(context.Background(), req.ReportLeaderOp); err != nil {
+		resp.Error = &proto.Error{Code: uint32(err.Code()), Msg: err.Message()}
+	}
+	data, err := resp.Marshal()
+	if err != nil {
+		panic(err)
+	}
+	return data
+}
+
 func (s *Server) isShutdown() bool {
 	s.mu.RLock()
 	defer s.mu.RUnlock()


Test patch:
diff --git a/server/api_test.go b/server/api_test.go
index 72272ae..19c6e76 100644
--- a/server/api_test.go
+++ b/server/api_test.go
@@ -302,9 +302,6 @@ func TestStreamPublishSubscribe(t *testing.T) {
 	ch1 := make(chan struct{})
 	ch2 := make(chan struct{})
 	err = client.Subscribe(context.Background(), name, func(msg *proto.Message, err error) {
-		if i == num+5 && err != nil {
-			return
-		}
 		require.NoError(t, err)
 		expect := expected[i]
 		assertMsg(t, expect, msg)
@@ -361,9 +358,6 @@ func TestStreamPublishSubscribe(t *testing.T) {
 	ch1 = make(chan struct{})
 	err = client2.Subscribe(context.Background(), name,
 		func(msg *proto.Message, err error) {
-			if i == num+5 && err != nil {
-				return
-			}
 			require.NoError(t, err)
 			expect := expected[i]
 			assertMsg(t, expect, msg)
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index 0fcab55..3df9601 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -15,10 +15,10 @@ import (
 
 var (
 	msgs = []*proto.Message{
-		&proto.Message{Value: []byte("one"), Timestamp: 1, LeaderEpoch: 42},
-		&proto.Message{Value: []byte("two"), Timestamp: 2, LeaderEpoch: 42},
-		&proto.Message{Value: []byte("three"), Timestamp: 3, LeaderEpoch: 42},
-		&proto.Message{Value: []byte("four"), Timestamp: 4, LeaderEpoch: 42},
+		{Value: []byte("one"), Timestamp: 1, LeaderEpoch: 42},
+		{Value: []byte("two"), Timestamp: 2, LeaderEpoch: 42},
+		{Value: []byte("three"), Timestamp: 3, LeaderEpoch: 42},
+		{Value: []byte("four"), Timestamp: 4, LeaderEpoch: 42},
 	}
 )
 
@@ -205,7 +205,7 @@ func TestCleanerDeleteLeaderEpochOffsets(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 1,
@@ -214,7 +214,7 @@ func TestCleanerDeleteLeaderEpochOffsets(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Value:       []byte(strconv.Itoa(i + 5)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 2,
@@ -223,7 +223,7 @@ func TestCleanerDeleteLeaderEpochOffsets(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Value:       []byte(strconv.Itoa(i + 10)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 3,
@@ -270,7 +270,7 @@ func TestCleanerReplaceLeaderEpochOffsets(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Key:         []byte("foo"),
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
@@ -280,7 +280,7 @@ func TestCleanerReplaceLeaderEpochOffsets(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Key:         []byte("bar"),
 			Value:       []byte(strconv.Itoa(i + 5)),
 			Timestamp:   time.Now().UnixNano(),
@@ -290,7 +290,7 @@ func TestCleanerReplaceLeaderEpochOffsets(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Key:         []byte("baz"),
 			Value:       []byte(strconv.Itoa(i + 10)),
 			Timestamp:   time.Now().UnixNano(),
@@ -390,7 +390,7 @@ func TestTruncate(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 1,
@@ -399,7 +399,7 @@ func TestTruncate(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Value:       []byte(strconv.Itoa(i + 5)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 2,
@@ -408,7 +408,7 @@ func TestTruncate(t *testing.T) {
 	}
 
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Value:       []byte(strconv.Itoa(i + 10)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 3,
@@ -442,7 +442,7 @@ func TestNotifyLEOMismatch(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 1,
@@ -452,7 +452,7 @@ func TestNotifyLEOMismatch(t *testing.T) {
 
 	// Get current log end offset and then add another message.
 	leo := l.NewestOffset()
-	_, err := l.Append([]*proto.Message{&proto.Message{
+	_, err := l.Append([]*proto.Message{{
 		Value:       []byte(strconv.Itoa(5)),
 		Timestamp:   time.Now().UnixNano(),
 		LeaderEpoch: 1,
@@ -482,7 +482,7 @@ func TestNotifyLEONewData(t *testing.T) {
 
 	// Add some messages.
 	for i := 0; i < 5; i++ {
-		_, err := l.Append([]*proto.Message{&proto.Message{
+		_, err := l.Append([]*proto.Message{{
 			Value:       []byte(strconv.Itoa(i)),
 			Timestamp:   time.Now().UnixNano(),
 			LeaderEpoch: 1,
@@ -504,7 +504,7 @@ func TestNotifyLEONewData(t *testing.T) {
 	}
 
 	// Add another message.
-	_, err := l.Append([]*proto.Message{&proto.Message{
+	_, err := l.Append([]*proto.Message{{
 		Value:       []byte(strconv.Itoa(5)),
 		Timestamp:   time.Now().UnixNano(),
 		LeaderEpoch: 1,
@@ -541,7 +541,7 @@ func TestNotifyLEOIdempotent(t *testing.T) {
 	require.Equal(t, ch1, ch2)
 }
 
-func setup(t require.TestingT) (*CommitLog, func()) {
+func setup(t require.TestingT) (*commitLog, func()) {
 	opts := Options{
 		Path:            tempDir(t),
 		MaxSegmentBytes: 6,
@@ -550,10 +550,10 @@ func setup(t require.TestingT) (*CommitLog, func()) {
 	return setupWithOptions(t, opts)
 }
 
-func setupWithOptions(t require.TestingT, opts Options) (*CommitLog, func()) {
+func setupWithOptions(t require.TestingT, opts Options) (*commitLog, func()) {
 	l, err := New(opts)
 	require.NoError(t, err)
-	return l, func() {
+	return l.(*commitLog), func() {
 		remove(t, opts.Path)
 	}
 }
diff --git a/server/commitlog/compact_cleaner_test.go b/server/commitlog/compact_cleaner_test.go
index 8c926f4..b409c88 100644
--- a/server/commitlog/compact_cleaner_test.go
+++ b/server/commitlog/compact_cleaner_test.go
@@ -23,8 +23,8 @@ type expectedMsg struct {
 
 // Ensure Compact is a no-op when there are no segments.
 func TestCompactCleanerNoSegments(t *testing.T) {
-	opts := CompactCleanerOptions{Name: "foo", Logger: noopLogger()}
-	cleaner := NewCompactCleaner(opts)
+	opts := compactCleanerOptions{Name: "foo", Logger: noopLogger()}
+	cleaner := newCompactCleaner(opts)
 	segments, epochCache, err := cleaner.Compact(0, nil)
 	require.NoError(t, err)
 	require.Nil(t, segments)
@@ -33,12 +33,12 @@ func TestCompactCleanerNoSegments(t *testing.T) {
 
 // Ensure Compact is a no-op when there is one segment.
 func TestCompactCleanerOneSegment(t *testing.T) {
-	opts := CompactCleanerOptions{Name: "foo", Logger: noopLogger()}
-	cleaner := NewCompactCleaner(opts)
+	opts := compactCleanerOptions{Name: "foo", Logger: noopLogger()}
+	cleaner := newCompactCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	expected := []*Segment{createSegment(t, dir, 0, 100)}
+	expected := []*segment{createSegment(t, dir, 0, 100)}
 	actual, epochCache, err := cleaner.Compact(0, expected)
 	require.NoError(t, err)
 	require.Equal(t, expected, actual)
@@ -58,16 +58,16 @@ func TestCompactCleaner(t *testing.T) {
 
 	// Append some messages.
 	entries := []keyValue{
-		keyValue{[]byte("foo"), []byte("first")},
-		keyValue{[]byte("bar"), []byte("first")},
-		keyValue{[]byte("foo"), []byte("second")},
-		keyValue{[]byte("foo"), []byte("third")},
-		keyValue{[]byte("bar"), []byte("second")},
-		keyValue{[]byte("baz"), []byte("first")},
-		keyValue{[]byte("baz"), []byte("second")},
-		keyValue{[]byte("qux"), []byte("first")},
-		keyValue{[]byte("foo"), []byte("fourth")},
-		keyValue{[]byte("baz"), []byte("third")},
+		{[]byte("foo"), []byte("first")},
+		{[]byte("bar"), []byte("first")},
+		{[]byte("foo"), []byte("second")},
+		{[]byte("foo"), []byte("third")},
+		{[]byte("bar"), []byte("second")},
+		{[]byte("baz"), []byte("first")},
+		{[]byte("baz"), []byte("second")},
+		{[]byte("qux"), []byte("first")},
+		{[]byte("foo"), []byte("fourth")},
+		{[]byte("baz"), []byte("third")},
 	}
 	appendToLog(t, l, entries, true)
 
@@ -75,11 +75,11 @@ func TestCompactCleaner(t *testing.T) {
 	require.NoError(t, l.Clean())
 
 	expected := []*expectedMsg{
-		&expectedMsg{Offset: 4, Msg: &proto.Message{Key: []byte("bar"), Value: []byte("second")}},
-		&expectedMsg{Offset: 7, Msg: &proto.Message{Key: []byte("qux"), Value: []byte("first")}},
-		&expectedMsg{Offset: 8, Msg: &proto.Message{Key: []byte("foo"), Value: []byte("fourth")}},
+		{Offset: 4, Msg: &proto.Message{Key: []byte("bar"), Value: []byte("second")}},
+		{Offset: 7, Msg: &proto.Message{Key: []byte("qux"), Value: []byte("first")}},
+		{Offset: 8, Msg: &proto.Message{Key: []byte("foo"), Value: []byte("fourth")}},
 		// This one is present because it's in the active segment.
-		&expectedMsg{Offset: 9, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("third")}},
+		{Offset: 9, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("third")}},
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -107,16 +107,16 @@ func TestCompactCleanerHW(t *testing.T) {
 
 	// Append some messages.
 	entries := []keyValue{
-		keyValue{[]byte("foo"), []byte("first")},
-		keyValue{[]byte("bar"), []byte("first")},
-		keyValue{[]byte("foo"), []byte("second")},
-		keyValue{[]byte("foo"), []byte("third")},
-		keyValue{[]byte("bar"), []byte("second")},
-		keyValue{[]byte("baz"), []byte("first")},
-		keyValue{[]byte("baz"), []byte("second")},
-		keyValue{[]byte("qux"), []byte("first")},
-		keyValue{[]byte("foo"), []byte("fourth")},
-		keyValue{[]byte("baz"), []byte("third")},
+		{[]byte("foo"), []byte("first")},
+		{[]byte("bar"), []byte("first")},
+		{[]byte("foo"), []byte("second")},
+		{[]byte("foo"), []byte("third")},
+		{[]byte("bar"), []byte("second")},
+		{[]byte("baz"), []byte("first")},
+		{[]byte("baz"), []byte("second")},
+		{[]byte("qux"), []byte("first")},
+		{[]byte("foo"), []byte("fourth")},
+		{[]byte("baz"), []byte("third")},
 	}
 	appendToLog(t, l, entries, false)
 	l.SetHighWatermark(5)
@@ -125,14 +125,14 @@ func TestCompactCleanerHW(t *testing.T) {
 	require.NoError(t, l.Clean())
 
 	expected := []*expectedMsg{
-		&expectedMsg{Offset: 3, Msg: &proto.Message{Key: []byte("foo"), Value: []byte("third")}},
-		&expectedMsg{Offset: 4, Msg: &proto.Message{Key: []byte("bar"), Value: []byte("second")}},
-		&expectedMsg{Offset: 5, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("first")}},
+		{Offset: 3, Msg: &proto.Message{Key: []byte("foo"), Value: []byte("third")}},
+		{Offset: 4, Msg: &proto.Message{Key: []byte("bar"), Value: []byte("second")}},
+		{Offset: 5, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("first")}},
 		// These are retained because they are after the HW.
-		&expectedMsg{Offset: 6, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("second")}},
-		&expectedMsg{Offset: 7, Msg: &proto.Message{Key: []byte("qux"), Value: []byte("first")}},
-		&expectedMsg{Offset: 8, Msg: &proto.Message{Key: []byte("foo"), Value: []byte("fourth")}},
-		&expectedMsg{Offset: 9, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("third")}},
+		{Offset: 6, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("second")}},
+		{Offset: 7, Msg: &proto.Message{Key: []byte("qux"), Value: []byte("first")}},
+		{Offset: 8, Msg: &proto.Message{Key: []byte("foo"), Value: []byte("fourth")}},
+		{Offset: 9, Msg: &proto.Message{Key: []byte("baz"), Value: []byte("third")}},
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -160,10 +160,10 @@ func TestCompactCleanerNoKeys(t *testing.T) {
 
 	// Append some messages.
 	entries := []keyValue{
-		keyValue{nil, []byte("first")},
-		keyValue{nil, []byte("second")},
-		keyValue{nil, []byte("third")},
-		keyValue{nil, []byte("fourth")},
+		{nil, []byte("first")},
+		{nil, []byte("second")},
+		{nil, []byte("third")},
+		{nil, []byte("fourth")},
 	}
 	appendToLog(t, l, entries, true)
 
@@ -171,10 +171,10 @@ func TestCompactCleanerNoKeys(t *testing.T) {
 	require.NoError(t, l.Clean())
 
 	expected := []*expectedMsg{
-		&expectedMsg{Offset: 0, Msg: &proto.Message{Value: []byte("first")}},
-		&expectedMsg{Offset: 1, Msg: &proto.Message{Value: []byte("second")}},
-		&expectedMsg{Offset: 2, Msg: &proto.Message{Value: []byte("third")}},
-		&expectedMsg{Offset: 3, Msg: &proto.Message{Value: []byte("fourth")}},
+		{Offset: 0, Msg: &proto.Message{Value: []byte("first")}},
+		{Offset: 1, Msg: &proto.Message{Value: []byte("second")}},
+		{Offset: 2, Msg: &proto.Message{Value: []byte("third")}},
+		{Offset: 3, Msg: &proto.Message{Value: []byte("fourth")}},
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -216,16 +216,16 @@ func TestCompactCleanerTruncateConcurrent(t *testing.T) {
 
 	// Append some messages.
 	entries := []keyValue{
-		keyValue{[]byte("foo"), []byte("first")},
-		keyValue{[]byte("bar"), []byte("first")},
-		keyValue{[]byte("foo"), []byte("second")},
-		keyValue{[]byte("foo"), []byte("third")},
-		keyValue{[]byte("bar"), []byte("second")},
-		keyValue{[]byte("baz"), []byte("first")},
-		keyValue{[]byte("baz"), []byte("second")},
-		keyValue{[]byte("qux"), []byte("first")},
-		keyValue{[]byte("foo"), []byte("fourth")},
-		keyValue{[]byte("baz"), []byte("third")},
+		{[]byte("foo"), []byte("first")},
+		{[]byte("bar"), []byte("first")},
+		{[]byte("foo"), []byte("second")},
+		{[]byte("foo"), []byte("third")},
+		{[]byte("bar"), []byte("second")},
+		{[]byte("baz"), []byte("first")},
+		{[]byte("baz"), []byte("second")},
+		{[]byte("qux"), []byte("first")},
+		{[]byte("foo"), []byte("fourth")},
+		{[]byte("baz"), []byte("third")},
 	}
 	appendToLog(t, l, entries, true)
 
@@ -299,7 +299,7 @@ func benchmarkClean(b *testing.B, segmentSize int64) {
 	}
 }
 
-func appendToLog(t *testing.T, l *CommitLog, entries []keyValue, commit bool) {
+func appendToLog(t *testing.T, l *commitLog, entries []keyValue, commit bool) {
 	for _, entry := range entries {
 		msg := &proto.Message{
 			Key:   entry.key,
diff --git a/server/commitlog/delete_cleaner_test.go b/server/commitlog/delete_cleaner_test.go
index 7145119..a61ca36 100644
--- a/server/commitlog/delete_cleaner_test.go
+++ b/server/commitlog/delete_cleaner_test.go
@@ -17,17 +17,17 @@ func noopLogger() logger.Logger {
 	return log
 }
 
-func createSegment(t require.TestingT, dir string, baseOffset, maxBytes int64) *Segment {
-	s, err := NewSegment(dir, baseOffset, maxBytes, false, "")
+func createSegment(t require.TestingT, dir string, baseOffset, maxBytes int64) *segment {
+	s, err := newSegment(dir, baseOffset, maxBytes, false, "")
 	require.NoError(t, err)
 	return s
 }
 
 // Ensure Clean is a no-op when there are no segments.
 func TestDeleteCleanerNoSegments(t *testing.T) {
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Bytes = 100
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	segments, err := cleaner.Clean(nil)
 	require.NoError(t, err)
 	require.Nil(t, segments)
@@ -35,12 +35,12 @@ func TestDeleteCleanerNoSegments(t *testing.T) {
 
 // Ensure Clean is a no-op when bytes and messages are 0.
 func TestDeleteCleanerNoRetentionSet(t *testing.T) {
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
-	cleaner := NewDeleteCleaner(opts)
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	expected := []*Segment{createSegment(t, dir, 0, 100)}
+	expected := []*segment{createSegment(t, dir, 0, 100)}
 	actual, err := cleaner.Clean(expected)
 	require.NoError(t, err)
 	require.Equal(t, expected, actual)
@@ -48,13 +48,13 @@ func TestDeleteCleanerNoRetentionSet(t *testing.T) {
 
 // Ensure Clean is a no-op when there is only one segment.
 func TestDeleteCleanerOneSegment(t *testing.T) {
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Bytes = 100
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	expected := []*Segment{createSegment(t, dir, 0, 100)}
+	expected := []*segment{createSegment(t, dir, 0, 100)}
 	actual, err := cleaner.Clean(expected)
 	require.NoError(t, err)
 	require.Equal(t, expected, actual)
@@ -62,13 +62,13 @@ func TestDeleteCleanerOneSegment(t *testing.T) {
 
 // Ensure Clean deletes segments to maintain the bytes limit.
 func TestDeleteCleanerBytes(t *testing.T) {
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Bytes = 100
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	segs := make([]*Segment, 5)
+	segs := make([]*segment, 5)
 	for i := 0; i < 5; i++ {
 		segs[i] = createSegment(t, dir, int64(i), 20)
 		writeToSegment(t, segs[i], int64(i), []byte("blah"))
@@ -83,13 +83,13 @@ func TestDeleteCleanerBytes(t *testing.T) {
 // Ensure Clean is a no-op when there are segments and a bytes limit but the
 // segments don't exceed the limit.
 func TestDeleteCleanerBytesBelowLimit(t *testing.T) {
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Bytes = 50
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	expected := make([]*Segment, 5)
+	expected := make([]*segment, 5)
 	for i := 0; i < 5; i++ {
 		expected[i] = createSegment(t, dir, int64(i), 20)
 	}
@@ -100,13 +100,13 @@ func TestDeleteCleanerBytesBelowLimit(t *testing.T) {
 
 // Ensure Clean deletes segments to maintain the messages limit.
 func TestDeleteCleanerMessages(t *testing.T) {
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Messages = 10
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	segs := make([]*Segment, 20)
+	segs := make([]*segment, 20)
 	for i := 0; i < 20; i++ {
 		segs[i] = createSegment(t, dir, int64(i), 20)
 		writeToSegment(t, segs[i], int64(i), []byte("blah"))
@@ -122,13 +122,13 @@ func TestDeleteCleanerMessages(t *testing.T) {
 // Ensure Clean is a no-op when there are segments and a messages limit but the
 // segments don't exceed the limit.
 func TestDeleteCleanerMessagesBelowLimit(t *testing.T) {
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Messages = 100
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	expected := make([]*Segment, 5)
+	expected := make([]*segment, 5)
 	for i := 0; i < 5; i++ {
 		expected[i] = createSegment(t, dir, int64(i), 20)
 	}
@@ -139,14 +139,14 @@ func TestDeleteCleanerMessagesBelowLimit(t *testing.T) {
 
 // Ensure Clean deletes segments to maintain the messages and bytes limits.
 func TestDeleteCleanerBytesMessages(t *testing.T) {
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Messages = 15
 	opts.Retention.Bytes = 240
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	segs := make([]*Segment, 20)
+	segs := make([]*segment, 20)
 	for i := 0; i < 20; i++ {
 		segs[i] = createSegment(t, dir, int64(i), 20)
 		writeToSegment(t, segs[i], int64(i), []byte("blah"))
@@ -169,17 +169,17 @@ func TestDeleteCleanerAge(t *testing.T) {
 		computeTTL = computeTTLBefore
 	}()
 
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Age = 100
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	segs := make([]*Segment, 20)
+	segs := make([]*segment, 20)
 	for i := 0; i < 20; i++ {
 		segs[i] = createSegment(t, dir, int64(i), 20)
-		ms, entries, err := NewMessageSetFromProto(int64(i), 0,
-			[]*proto.Message{&proto.Message{Timestamp: int64(i * 10)}})
+		ms, entries, err := newMessageSetFromProto(int64(i), 0,
+			[]*proto.Message{{Timestamp: int64(i * 10)}})
 		require.NoError(t, err)
 		require.NoError(t, segs[i].WriteMessageSet(ms, entries))
 	}
@@ -202,17 +202,17 @@ func TestDeleteCleanerMessagesBelowAgeLimit(t *testing.T) {
 		computeTTL = computeTTLBefore
 	}()
 
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Age = 50
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
-	expected := make([]*Segment, 5)
+	expected := make([]*segment, 5)
 	for i := 0; i < 5; i++ {
 		expected[i] = createSegment(t, dir, int64(i), 20)
-		ms, entries, err := NewMessageSetFromProto(int64(i), 0,
-			[]*proto.Message{&proto.Message{Timestamp: int64(i * 10)}})
+		ms, entries, err := newMessageSetFromProto(int64(i), 0,
+			[]*proto.Message{{Timestamp: int64(i * 10)}})
 		require.NoError(t, err)
 		require.NoError(t, expected[i].WriteMessageSet(ms, entries))
 	}
@@ -224,9 +224,9 @@ func TestDeleteCleanerMessagesBelowAgeLimit(t *testing.T) {
 // Ensure Clean correctly calculates the number of messages in the log when
 // it's been compacted.
 func TestDeleteCleanerMessagesCompacted(t *testing.T) {
-	opts := DeleteCleanerOptions{Name: "foo", Logger: noopLogger()}
+	opts := deleteCleanerOptions{Name: "foo", Logger: noopLogger()}
 	opts.Retention.Messages = 10
-	cleaner := NewDeleteCleaner(opts)
+	cleaner := newDeleteCleaner(opts)
 	dir := tempDir(t)
 	defer remove(t, dir)
 
@@ -241,14 +241,14 @@ func TestDeleteCleanerMessagesCompacted(t *testing.T) {
 	writeToSegment(t, seg2, 14, []byte("blah"))
 	writeToSegment(t, seg2, 15, []byte("blah"))
 
-	segs := []*Segment{seg1, seg2}
+	segs := []*segment{seg1, seg2}
 	actual, err := cleaner.Clean(segs)
 
 	require.NoError(t, err)
 	require.Len(t, actual, 2)
 
 	// Ensure no messages were actually deleted.
-	ss := NewSegmentScanner(actual[0])
+	ss := newSegmentScanner(actual[0])
 	_, entry, err := ss.Scan()
 	require.NoError(t, err)
 	require.Equal(t, int64(2), entry.Offset)
@@ -261,7 +261,7 @@ func TestDeleteCleanerMessagesCompacted(t *testing.T) {
 	_, _, err = ss.Scan()
 	require.Error(t, err)
 
-	ss = NewSegmentScanner(actual[1])
+	ss = newSegmentScanner(actual[1])
 	_, entry, err = ss.Scan()
 	require.NoError(t, err)
 	require.Equal(t, int64(13), entry.Offset)
@@ -275,10 +275,10 @@ func TestDeleteCleanerMessagesCompacted(t *testing.T) {
 	require.Error(t, err)
 }
 
-func writeToSegment(t *testing.T, seg *Segment, offset int64, data []byte) {
-	ms, entries, err := NewMessageSetFromProto(int64(offset), seg.Position(),
+func writeToSegment(t *testing.T, seg *segment, offset int64, data []byte) {
+	ms, entries, err := newMessageSetFromProto(int64(offset), seg.Position(),
 		[]*proto.Message{
-			&proto.Message{
+			{
 				Timestamp:   time.Now().UnixNano(),
 				LeaderEpoch: 42,
 				Value:       data,
diff --git a/server/commitlog/index_test.go b/server/commitlog/index_test.go
index dfe361d..92c9a85 100644
--- a/server/commitlog/index_test.go
+++ b/server/commitlog/index_test.go
@@ -12,7 +12,7 @@ func TestIndexInitialSize(t *testing.T) {
 	defer os.RemoveAll(dir)
 
 	// Create a new index.
-	idx, err := NewIndex(options{path: dir + "test.idx"})
+	idx, err := newIndex(options{path: dir + "test.idx"})
 	require.NoError(t, err)
 	entry, err := idx.InitializePosition()
 	require.NoError(t, err)
@@ -33,14 +33,14 @@ func TestIndexExistingSize(t *testing.T) {
 	defer os.RemoveAll(dir)
 
 	// Create a new index.
-	idx, err := NewIndex(options{path: dir + "test.idx"})
+	idx, err := newIndex(options{path: dir + "test.idx"})
 	require.NoError(t, err)
-	entry, err := idx.InitializePosition()
+	e, err := idx.InitializePosition()
 	require.NoError(t, err)
-	require.Nil(t, entry)
+	require.Nil(t, e)
 
 	// Write some entries.
-	err = idx.writeEntries([]*Entry{
+	err = idx.writeEntries([]*entry{
 		{}, {}, {}, {},
 	})
 	require.NoError(t, err)
@@ -50,7 +50,7 @@ func TestIndexExistingSize(t *testing.T) {
 	require.NoError(t, err)
 
 	// Reopen the index and verify the size and position are correct.
-	idx, err = NewIndex(options{path: dir + "test.idx"})
+	idx, err = newIndex(options{path: dir + "test.idx"})
 	require.NoError(t, err)
 	require.Equal(t, idx.size, int64(entryWidth*4))
 	require.Equal(t, idx.size, idx.position)
@@ -61,19 +61,19 @@ func TestIndexExpansion(t *testing.T) {
 	defer os.RemoveAll(dir)
 
 	// Create an index with enough room for a single entry.
-	idx, err := NewIndex(options{path: dir + "test.idx", bytes: entryWidth})
+	idx, err := newIndex(options{path: dir + "test.idx", bytes: entryWidth})
 	require.NoError(t, err)
-	entry, err := idx.InitializePosition()
+	e, err := idx.InitializePosition()
 	require.NoError(t, err)
-	require.Nil(t, entry)
+	require.Nil(t, e)
 
 	// Write two entries.
-	writeEntry := Entry{Offset: 123, Timestamp: 456, Position: 789, Size: 987}
-	err = idx.writeEntries([]*Entry{{}, &writeEntry})
+	writeEntry := entry{Offset: 123, Timestamp: 456, Position: 789, Size: 987}
+	err = idx.writeEntries([]*entry{{}, &writeEntry})
 	require.NoError(t, err)
 
 	// Check the second entry can be retrieved.
-	var readEntry Entry
+	var readEntry entry
 	err = idx.ReadEntryAtLogOffset(&readEntry, 1)
 	require.NoError(t, err)
 	require.Equal(t, writeEntry, readEntry)
diff --git a/server/commitlog/segment_test.go b/server/commitlog/segment_test.go
index 9cac33b..ac10c6c 100644
--- a/server/commitlog/segment_test.go
+++ b/server/commitlog/segment_test.go
@@ -17,7 +17,7 @@ func TestSegmentCheckSplitFull(t *testing.T) {
 	s := createSegment(t, dir, 0, 10)
 	require.False(t, s.CheckSplit(1))
 
-	_, err := s.write(make([]byte, 10), []*Entry{&Entry{}})
+	_, err := s.write(make([]byte, 10), []*entry{{}})
 	require.NoError(t, err)
 	require.True(t, s.CheckSplit(1))
 }
@@ -46,7 +46,7 @@ func TestSegmentCheckSplitNotFull(t *testing.T) {
 	}()
 
 	s := createSegment(t, dir, 0, 10)
-	_, err := s.write(make([]byte, 5), []*Entry{&Entry{}})
+	_, err := s.write(make([]byte, 5), []*entry{{}})
 	require.NoError(t, err)
 	s.firstWriteTime = 1
 	require.False(t, s.CheckSplit(5))
@@ -66,7 +66,7 @@ func TestSegmentCheckSplitLogRollTimeExceeded(t *testing.T) {
 	}()
 
 	s := createSegment(t, dir, 0, 10)
-	_, err := s.write(make([]byte, 5), []*Entry{&Entry{}})
+	_, err := s.write(make([]byte, 5), []*entry{{}})
 	require.NoError(t, err)
 	s.firstWriteTime = 1
 	require.True(t, s.CheckSplit(1))
diff --git a/server/commitlog/util_test.go b/server/commitlog/util_test.go
index 2c09aa9..ffc04df 100644
--- a/server/commitlog/util_test.go
+++ b/server/commitlog/util_test.go
@@ -7,12 +7,12 @@ import (
 )
 
 func TestFindSegment(t *testing.T) {
-	segments := []*Segment{
-		&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},
+	segments := []*segment{
+		{BaseOffset: 0, lastOffset: 9},
+		{BaseOffset: 10, lastOffset: 19},
+		{BaseOffset: 20, lastOffset: 29},
+		{BaseOffset: 30, lastOffset: 39},
+		{BaseOffset: 40, lastOffset: 49},
 	}
 	seg, idx := findSegment(segments, 0)
 	require.Equal(t, 0, idx)
@@ -36,12 +36,12 @@ func TestFindSegment(t *testing.T) {
 }
 
 func TestFindSegmentByBaseOffset(t *testing.T) {
-	segments := []*Segment{
-		&Segment{BaseOffset: 0},
-		&Segment{BaseOffset: 10},
-		&Segment{BaseOffset: 20},
-		&Segment{BaseOffset: 30},
-		&Segment{BaseOffset: 40},
+	segments := []*segment{
+		{BaseOffset: 0},
+		{BaseOffset: 10},
+		{BaseOffset: 20},
+		{BaseOffset: 30},
+		{BaseOffset: 40},
 	}
 	require.Equal(t, segments[0], findSegmentByBaseOffset(segments, 0))
 	require.Equal(t, segments[1], findSegmentByBaseOffset(segments, 1))
diff --git a/server/conf/lex_test.go b/server/conf/lex_test.go
index 332e522..8558605 100644
--- a/server/conf/lex_test.go
+++ b/server/conf/lex_test.go
@@ -906,7 +906,7 @@ func TestBlockStringEOF(t *testing.T) {
 		{itemKey, "numbers", 2},
 		{itemString, "\n1234567890\n", 4},
 	}
-	blockbytes := []byte(blockexample[0:len(blockexample)-1])
+	blockbytes := []byte(blockexample[0 : len(blockexample)-1])
 	blockbytes = append(blockbytes, 0)
 	lx := lex(string(blockbytes))
 	expect(t, lx, expectedItems)
diff --git a/server/replicator_test.go b/server/replicator_test.go
index 4254cf9..46c3373 100644
--- a/server/replicator_test.go
+++ b/server/replicator_test.go
@@ -12,7 +12,6 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/require"
 
-	"github.com/liftbridge-io/liftbridge/server/commitlog"
 	internal "github.com/liftbridge-io/liftbridge/server/proto"
 )
 
@@ -232,7 +231,7 @@ func TestCommitOnISRShrink(t *testing.T) {
 	defer s3.Stop()
 
 	servers := []*Server{s1, s2, s3}
-	leader := getMetadataLeader(t, 10*time.Second, servers...)
+	getMetadataLeader(t, 10*time.Second, servers...)
 
 	client, err := lift.Connect([]string{"localhost:5050", "localhost:5051", "localhost:5052"})
 	require.NoError(t, err)
@@ -246,7 +245,7 @@ func TestCommitOnISRShrink(t *testing.T) {
 	require.NoError(t, err)
 
 	// Kill a stream follower.
-	leader = getPartitionLeader(t, 10*time.Second, name, 0, servers...)
+	leader := getPartitionLeader(t, 10*time.Second, name, 0, servers...)
 	var follower *Server
 	for i, server := range servers {
 		if server != leader {
@@ -261,7 +260,8 @@ func TestCommitOnISRShrink(t *testing.T) {
 	// shrinks.
 	gotAck := make(chan error)
 	go func() {
-		ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
 		_, err := client.Publish(ctx, subject, []byte("hello"), lift.AckPolicyAll())
 		gotAck <- err
 	}()
@@ -306,7 +306,7 @@ func TestAckPolicyLeader(t *testing.T) {
 	defer s3.Stop()
 
 	servers := []*Server{s1, s2, s3}
-	leader := getMetadataLeader(t, 10*time.Second, servers...)
+	getMetadataLeader(t, 10*time.Second, servers...)
 
 	client, err := lift.Connect([]string{"localhost:5050", "localhost:5051", "localhost:5052"})
 	require.NoError(t, err)
@@ -320,7 +320,7 @@ func TestAckPolicyLeader(t *testing.T) {
 	require.NoError(t, err)
 
 	// Kill a stream follower.
-	leader = getPartitionLeader(t, 10*time.Second, name, 0, servers...)
+	leader := getPartitionLeader(t, 10*time.Second, name, 0, servers...)
 	var follower *Server
 	for i, server := range servers {
 		if server != leader {
@@ -334,7 +334,8 @@ func TestAckPolicyLeader(t *testing.T) {
 	// Publish message to stream. This should not get committed until the ISR
 	// shrinks, but an ack should still be received immediately since
 	// AckPolicy_LEADER is set (default AckPolicy).
-	ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	cid := "cid"
 	ack, err := client.Publish(ctx, subject, []byte("hello"),
 		lift.CorrelationID(cid))
@@ -381,7 +382,8 @@ func TestCommitOnRestart(t *testing.T) {
 	// Publish some messages.
 	num := 5
 	for i := 0; i < num; i++ {
-		ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
 		_, err = client.Publish(ctx, subject, []byte("hello"), lift.AckPolicyAll())
 		require.NoError(t, err)
 	}
@@ -400,7 +402,8 @@ func TestCommitOnRestart(t *testing.T) {
 
 	// Publish some more messages.
 	for i := 0; i < num; i++ {
-		ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
 		_, err = client.Publish(ctx, subject, []byte("hello"))
 		require.NoError(t, err)
 	}
@@ -504,10 +507,12 @@ func TestTruncateFastLeaderElection(t *testing.T) {
 	require.NoError(t, err)
 
 	// Publish two messages.
-	ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	_, err = client.Publish(ctx, subject, []byte("hello"), lift.AckPolicyAll())
 	require.NoError(t, err)
-	ctx, _ = context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	_, err = client.Publish(ctx, subject, []byte("world"), lift.AckPolicyAll())
 	require.NoError(t, err)
 
@@ -538,13 +543,13 @@ func TestTruncateFastLeaderElection(t *testing.T) {
 	partition1 := follower1.metadata.GetPartition(name, 0)
 	require.NotNil(t, partition1)
 	require.NoError(t, partition1.stopFollowing())
-	partition1.log.(*commitlog.CommitLog).OverrideHighWatermark(0)
+	partition1.log.OverrideHighWatermark(0)
 
 	// Stop second follower's replication and reset HW.
 	partition2 := follower2.metadata.GetPartition(name, 0)
 	require.NotNil(t, partition2)
 	require.NoError(t, partition2.stopFollowing())
-	partition2.log.(*commitlog.CommitLog).OverrideHighWatermark(0)
+	partition2.log.OverrideHighWatermark(0)
 
 	var (
 		follower1Config *Config
@@ -636,10 +641,12 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 	require.NoError(t, err)
 
 	// Publish two messages.
-	ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	_, err = client.Publish(ctx, subject, []byte("hello"))
 	require.NoError(t, err)
-	ctx, _ = context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	_, err = client.Publish(ctx, subject, []byte("world"))
 	require.NoError(t, err)
 
@@ -672,7 +679,7 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 	partition1.mu.Lock()
 	require.NoError(t, partition1.stopFollowing())
 	partition1.mu.Unlock()
-	partition1.log.(*commitlog.CommitLog).OverrideHighWatermark(0)
+	partition1.log.OverrideHighWatermark(0)
 	partition1.truncateToHW()
 
 	// Stop second follower's replication and reset HW.
@@ -681,7 +688,7 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 	partition2.mu.Lock()
 	require.NoError(t, partition2.stopFollowing())
 	partition2.mu.Unlock()
-	partition2.log.(*commitlog.CommitLog).OverrideHighWatermark(0)
+	partition2.log.OverrideHighWatermark(0)
 	partition2.truncateToHW()
 
 	var (
@@ -730,11 +737,13 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 	waitForISR(t, 10*time.Second, name, 0, 2, follower1, follower2)
 
 	// Publish new messages.
-	ctx, _ = context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	_, err = client.Publish(ctx, subject, []byte("goodnight"))
 	require.NoError(t, err)
 
-	ctx, _ = context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	_, err = client.Publish(ctx, subject, []byte("moon"))
 	require.NoError(t, err)