Add WithSyncInterval option for delayed BBolt disk flushes

bbolt/bbolt.go

@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"os"
 	"path"
+	"sync/atomic"
 	"time"
 
 	"github.com/nuts-foundation/go-stoabs"
@@ -53,6 +54,11 @@ func CreateBBoltStore(filePath string, opts ...stoabs.Option) (stoabs.KVStore, e
 		bboltOpts.NoSync = true
 		bboltOpts.NoFreelistSync = true
 		bboltOpts.NoGrowSync = true
+	} else if cfg.DelayedSync() {
+		// Disable bbolt's per-commit syncs; we will sync on a timer instead.
+		bboltOpts.NoSync = true
+		bboltOpts.NoFreelistSync = true
+		bboltOpts.NoGrowSync = true
 	}
 	return createBBoltStore(filePath, &bboltOpts, cfg)
 }
@@ -88,23 +94,44 @@ func createBBoltStore(filePath string, options *bbolt.Options, cfg stoabs.Config
 
 // Wrap creates a KVStore using an existing bbolt.db
 func Wrap(db *bbolt.DB, cfg stoabs.Config) stoabs.KVStore {
-	return &store{
+	s := &store{
 		db:   db,
 		cfg:  cfg,
 		log:  cfg.Log,
 		lock: &util.ContextRWLocker{},
 	}
+	if cfg.DelayedSync() {
+		ctx, cancelCtx := context.WithCancel(context.Background())
+		s.cancelCtx = cancelCtx
+		s.startSync(ctx, cfg.SyncInterval)
+	}
+	return s
 }
 
 type store struct {
-	db   *bbolt.DB
-	log  *logrus.Logger
-	lock *util.ContextRWLocker
-	cfg  stoabs.Config
+	db        *bbolt.DB
+	log       *logrus.Logger
+	lock      *util.ContextRWLocker
+	cfg       stoabs.Config
+	mustSync  atomic.Bool
+	cancelCtx context.CancelFunc
 }
 
 func (b *store) Close(ctx context.Context) error {
-	err := util.CallWithTimeout(ctx, b.db.Close, func() {
+	err := util.CallWithTimeout(ctx, func() error {
+		// Stop the background sync goroutine before flushing so it cannot
+		// race with our explicit Sync call below.
+		if b.cancelCtx != nil {
+			b.cancelCtx()
+		}
+		// Flush any writes that were committed since the last periodic sync.
+		if b.mustSync.Swap(false) {
+			if err := b.db.Sync(); err != nil {
+				return fmt.Errorf("could not flush to disk: %w", err)
+			}
+		}
+		return b.db.Close()
+	}, func() {
 		b.log.Error("Closing of BBolt store timed out, store may not shut down correctly.")
 	})
 	if err != nil {
@@ -199,11 +226,35 @@ func (b *store) doTX(ctx context.Context, fn func(tx *bbolt.Tx) error, writable
 		return util.WrapError(stoabs.ErrCommitFailed, err)
 	}
 
+	// Mark that there are unflushed writes, so the sync goroutine (or Close) will flush them.
+	if b.cfg.DelayedSync() {
+		b.mustSync.Store(true)
+	}
+
 	unlock()
 	stoabs.AfterCommitOption{}.Invoke(opts)
 	return nil
 }
 
+func (b *store) startSync(ctx context.Context, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	go func() {
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				if b.mustSync.Swap(false) {
+					if err := b.db.Sync(); err != nil {
+						b.log.WithError(err).Error("Failed to sync BBolt store to disk")
+					}
+				}
+			}
+		}
+	}()
+}
+
 func rollbackTX(dbTX *bbolt.Tx, log *logrus.Logger) {
 	err := dbTX.Rollback()
 	if err != nil {

bbolt/bbolt_test.go

@@ -137,6 +137,87 @@ func TestBBolt_CreateBBoltStore(t *testing.T) {
 	})
 }
 
+func TestBBolt_SyncInterval(t *testing.T) {
+	ctx := context.Background()
+	const numEntries = 100
+	data := make([]byte, 1024)
+
+	t.Run("data is flushed on close when dirty", func(t *testing.T) {
+		dbPath := path.Join(util.TestDirectory(t), "bbolt.db")
+		store, _ := CreateBBoltStore(dbPath, stoabs.WithSyncInterval(time.Second))
+
+		for i := 0; i < numEntries; i++ {
+			err := store.WriteShelf(ctx, shelf, func(writer stoabs.Writer) error {
+				return writer.Put(stoabs.BytesKey(fmt.Sprintf("%d", i)), data)
+			})
+			if !assert.NoError(t, err) {
+				return
+			}
+		}
+		assert.NoError(t, store.Close(ctx))
+
+		// Reopen and verify all entries survived the close
+		store, _ = CreateBBoltStore(dbPath, stoabs.WithSyncInterval(time.Second))
+		defer store.Close(ctx)
+		err := store.ReadShelf(ctx, shelf, func(reader stoabs.Reader) error {
+			assert.Equal(t, uint(numEntries), reader.Stats().NumEntries)
+			return nil
+		})
+		assert.NoError(t, err)
+	})
+
+	t.Run("data is flushed periodically", func(t *testing.T) {
+		dbPath := path.Join(util.TestDirectory(t), "bbolt.db")
+		kvStore, _ := CreateBBoltStore(dbPath, stoabs.WithSyncInterval(50*time.Millisecond))
+
+		err := kvStore.WriteShelf(ctx, shelf, func(writer stoabs.Writer) error {
+			return writer.Put(stoabs.BytesKey("key"), data)
+		})
+		assert.NoError(t, err)
+
+		// Wait for the periodic sync to fire
+		util.WaitFor(t, func() (bool, error) {
+			return !kvStore.(*store).mustSync.Load(), nil
+		}, time.Second, "timed out waiting for periodic sync")
+
+		assert.NoError(t, kvStore.Close(ctx))
+	})
+}
+
+func BenchmarkBBolt_Flush(b *testing.B) {
+	ctx := context.Background()
+	const numEntries = 10
+	data := make([]byte, 1024)
+
+	b.Run("sync on every commit", func(b *testing.B) {
+		dbPath := path.Join(b.TempDir(), "bbolt.db")
+		kvStore, _ := CreateBBoltStore(dbPath)
+		b.ResetTimer()
+		for x := 0; x < b.N; x++ {
+			for i := 0; i < numEntries; i++ {
+				_ = kvStore.WriteShelf(ctx, shelf, func(writer stoabs.Writer) error {
+					return writer.Put(stoabs.BytesKey(fmt.Sprintf("%d", i)), data)
+				})
+			}
+		}
+		_ = kvStore.Close(ctx)
+	})
+
+	b.Run("sync at interval", func(b *testing.B) {
+		dbPath := path.Join(b.TempDir(), "bbolt.db")
+		kvStore, _ := CreateBBoltStore(dbPath, stoabs.WithSyncInterval(time.Second))
+		b.ResetTimer()
+		for x := 0; x < b.N; x++ {
+			for i := 0; i < numEntries; i++ {
+				_ = kvStore.WriteShelf(ctx, shelf, func(writer stoabs.Writer) error {
+					return writer.Put(stoabs.BytesKey(fmt.Sprintf("%d", i)), data)
+				})
+			}
+		}
+		_ = kvStore.Close(ctx)
+	})
+}
+
 func TestBBolt_Close(t *testing.T) {
 	ctx := context.Background()
 	var bytesKey = stoabs.BytesKey([]byte{1, 2, 3})

store.go

@@ -98,9 +98,20 @@ type Option func(config *Config)
 type Config struct {
 	Log                *logrus.Logger
 	NoSync             bool
+	// SyncInterval specifies how often pending writes should be flushed to disk, instead of flushing after every
+	// transaction commit. This can significantly improve write throughput at the cost of potentially losing the last
+	// interval's worth of writes on an unclean shutdown. If zero, writes are flushed after every commit.
+	// Support depends on the underlying database. Ignored when NoSync is true.
+	SyncInterval       time.Duration
 	LockAcquireTimeout time.Duration
 }
 
+// DelayedSync reports whether writes should be buffered and flushed on a timer
+// rather than synced after every commit.
+func (c Config) DelayedSync() bool {
+	return c.SyncInterval > 0 && !c.NoSync
+}
+
 // DefaultConfig returns the default configuration.
 func DefaultConfig() Config {
 	return Config{
@@ -124,6 +135,16 @@ func WithNoSync() Option {
 	}
 }
 
+// WithSyncInterval specifies that the database should flush pending writes to disk at most at the given interval,
+// rather than after every transaction commit. This can significantly improve write throughput at the cost of
+// potentially losing the last interval's worth of writes on an unclean shutdown.
+// Support depends on the underlying database.
+func WithSyncInterval(interval time.Duration) Option {
+	return func(config *Config) {
+		config.SyncInterval = interval
+	}
+}
+
 // WithLogger overrides the default logger.
 func WithLogger(log *logrus.Logger) Option {
 	return func(config *Config) {