From e71a83b5ec8e4b414e4e30ea608bca423cc8bf05 Mon Sep 17 00:00:00 2001
From: Injun Song <ijsong@gmail.com>
Date: Fri, 28 Oct 2022 10:52:24 +0900
Subject: [PATCH] refactor(storage): do not write a commit context for each
 commit

Previously, the log stream wrote a commit context for each commit. However, to fix #125, we no
longer need to persist in a sequence of commit contexts. So this patch removes lines that keep all
commit context.

Updates #125
---
 internal/storage/storage.go      | 41 ------------------------
 internal/storage/storage_test.go | 54 --------------------------------
 2 files changed, 95 deletions(-)

diff --git a/internal/storage/storage.go b/internal/storage/storage.go
index 5ce2e4ea5..0c2aca09d 100644
--- a/internal/storage/storage.go
+++ b/internal/storage/storage.go
@@ -93,10 +93,6 @@ func (s *Storage) NewCommitBatch(cc CommitContext) (*CommitBatch, error) {
 		_ = cb.Close()
 		return nil, err
 	}
-	if err := cb.batch.Set(encodeCommitContextKeyInternal(cc, cb.cck), nil, nil); err != nil {
-		_ = cb.Close()
-		return nil, err
-	}
 	return cb, nil
 }
 
@@ -184,43 +180,6 @@ func (s *Storage) ReadCommitContext() (cc CommitContext, err error) {
 	return decodeCommitContext(buf), nil
 }
 
-// CommitContextOf looks up a commit context that contains the log entry for the argument glsn.
-func (s *Storage) CommitContextOf(glsn types.GLSN) (cc CommitContext, err error) {
-	cck := make([]byte, commitContextKeyLength)
-	cck = encodeCommitContextKeyInternal(CommitContext{HighWatermark: glsn}, cck)
-	it := s.db.NewIter(&pebble.IterOptions{
-		LowerBound: cck,
-		UpperBound: []byte{commitContextKeySentinelPrefix},
-	})
-	defer func() {
-		_ = it.Close()
-	}()
-	if it.First() {
-		cc = decodeCommitContextKey(it.Key())
-		if cc.CommittedGLSNBegin <= glsn && glsn < cc.CommittedGLSNEnd {
-			return cc, nil
-		}
-	}
-	return cc, ErrNoCommitContext
-}
-
-// NextCommitContextOf returns the next commit context after the argument cc if exists.
-func (s *Storage) NextCommitContextOf(cc CommitContext) (next CommitContext, err error) {
-	cck := make([]byte, commitContextKeyLength)
-	cck = encodeCommitContextKeyInternal(cc, cck)
-	it := s.db.NewIter(&pebble.IterOptions{
-		LowerBound: cck,
-		UpperBound: []byte{commitContextKeySentinelPrefix},
-	})
-	defer func() {
-		_ = it.Close()
-	}()
-	if !it.First() || !decodeCommitContextKey(it.Key()).Equal(cc) || !it.Next() {
-		return next, ErrNoCommitContext
-	}
-	return decodeCommitContextKey(it.Key()), nil
-}
-
 // Trim deletes log entries whose GLSNs are less than or equal to the argument
 // glsn. Internally, it removes records for both data and commits but does not
 // remove the commit context.
diff --git a/internal/storage/storage_test.go b/internal/storage/storage_test.go
index 45304aa38..6198c7618 100644
--- a/internal/storage/storage_test.go
+++ b/internal/storage/storage_test.go
@@ -723,60 +723,6 @@ func TestStorageReadRecoveryPoints_InconsistentWriteCommit(t *testing.T) {
 	})
 }
 
-func TestStorage_CommitContextOf(t *testing.T) {
-	testStorage(t, func(t testing.TB, stg *Storage) {
-		cc1 := CommitContext{
-			Version:            1,
-			HighWatermark:      10,
-			CommittedGLSNBegin: 8,
-			CommittedGLSNEnd:   11,
-			CommittedLLSNBegin: 1,
-		}
-		cc2 := CommitContext{
-			Version:            2,
-			HighWatermark:      22,
-			CommittedGLSNBegin: 21,
-			CommittedGLSNEnd:   23,
-			CommittedLLSNBegin: 4,
-		}
-		cb, err := stg.NewCommitBatch(cc1)
-		assert.NoError(t, err)
-		assert.NoError(t, cb.Apply())
-		assert.NoError(t, cb.Close())
-
-		cb, err = stg.NewCommitBatch(cc2)
-		assert.NoError(t, err)
-		assert.NoError(t, cb.Apply())
-		assert.NoError(t, cb.Close())
-
-		for glsn := types.GLSN(0); glsn < cc1.CommittedGLSNBegin; glsn++ {
-			_, err := stg.CommitContextOf(glsn)
-			assert.ErrorIs(t, err, ErrNoCommitContext)
-		}
-		for glsn := cc1.CommittedGLSNBegin; glsn < cc1.CommittedGLSNEnd; glsn++ {
-			cc, err := stg.CommitContextOf(glsn)
-			assert.NoError(t, err)
-			assert.Equal(t, cc1, cc)
-
-			cc, err = stg.NextCommitContextOf(cc)
-			assert.NoError(t, err)
-			assert.Equal(t, cc2, cc)
-		}
-		for glsn := cc1.CommittedGLSNEnd; glsn < cc2.CommittedGLSNBegin; glsn++ {
-			_, err := stg.CommitContextOf(glsn)
-			assert.ErrorIs(t, err, ErrNoCommitContext)
-		}
-		for glsn := cc2.CommittedGLSNBegin; glsn < cc2.CommittedGLSNEnd; glsn++ {
-			cc, err := stg.CommitContextOf(glsn)
-			assert.NoError(t, err)
-			assert.Equal(t, cc2, cc)
-
-			_, err = stg.NextCommitContextOf(cc)
-			assert.ErrorIs(t, err, ErrNoCommitContext)
-		}
-	})
-}
-
 func TestStorage_TrimWhenNoLogEntry(t *testing.T) {
 	tcs := []struct {
 		name  string