Multiple fixes

 - Timer efficiency improvements
 - Timer based change log cleanup
 - Improvements to documentation
 - Removing redundant values from replication event
 - Improving replicator to only create streams if they are missing, this
will let admins operations persist on jetstreams in case of restart

README.md

@@ -13,7 +13,7 @@ as a side car letting you build replication cluster without making any changes t
 application code, and allows you to keep using to your SQLite database file. 
 
 ## Dependencies
-Starting 0.4+ Marmot depends on [nats-server](https://nats.io/download/).
+Starting 0.4+ Marmot depends on [nats-server](https://nats.io/download/) with JetStream support.
 
 ## Production status
 

db/change_log.go

@@ -89,12 +89,15 @@ func (conn *SqliteStreamDB) DeleteChangeLog(event *ChangeLogEvent) (bool, error)
 	return count > 0, nil
 }
 
-func (conn *SqliteStreamDB) CleanupChangeLogs() (int64, error) {
+func (conn *SqliteStreamDB) CleanupChangeLogs(beforeTime time.Time) (int64, error) {
 	total := int64(0)
 	for name := range conn.watchTablesSchema {
 		metaTableName := conn.metaTable(name, changeLogName)
 		rs, err := conn.Delete(metaTableName).
-			Where(goqu.Ex{"state": Published}).
+			Where(
+				goqu.C("state").Eq(Published),
+				goqu.C("created_at").Lte(beforeTime.UnixMilli()),
+			).
 			Prepared(true).
 			Executor().
 			Exec()
@@ -199,6 +202,8 @@ func (conn *SqliteStreamDB) watchChanges(path string) {
 
 	errShm := watcher.Add(shmPath)
 	errWal := watcher.Add(walPath)
+	changeLogTicker := time.NewTicker(time.Millisecond * 100)
+	updateTime := time.Now()
 
 	for {
 		select {
@@ -210,8 +215,11 @@ func (conn *SqliteStreamDB) watchChanges(path string) {
 			if ev.Op != fsnotify.Chmod {
 				conn.publishChangeLog()
 			}
-		case <-time.After(time.Millisecond * 500):
-			conn.publishChangeLog()
+		case t := <-changeLogTicker.C:
+			if t.After(updateTime) {
+				conn.publishChangeLog()
+			}
+
 			if errShm != nil {
 				errShm = watcher.Add(shmPath)
 			}
@@ -220,6 +228,8 @@ func (conn *SqliteStreamDB) watchChanges(path string) {
 				errWal = watcher.Add(walPath)
 			}
 		}
+
+		updateTime = time.Now()
 	}
 }
 

db/change_log_event.go

@@ -75,7 +75,7 @@ func (e *ChangeLogEvent) getSortedPKColumns() []string {
 
 	tablePKColumnsLock.Lock()
 	defer tablePKColumnsLock.Unlock()
-	tablePKColumnsCache[e.TableName] = pkColumns
 
+	tablePKColumnsCache[e.TableName] = pkColumns
 	return pkColumns
 }

docs/overview.md

@@ -30,9 +30,6 @@ Yes it will require additional storage to old/new values from triggers. But righ
 Ask marmot to remove hooks and log tables by:
 `marmot -db-path /path/to/your/db.db -cleanup`
 
-Cleanup the Raft logs and snapshot by:
-`rm -rf /tmp/raft`
-
 ### How would many shards should I have?
 
 It depends on your usecase and what problem you are solving for. In a typical setting you should not need more than couple of dozen shards. While read scaling won't be a problem, your write throughput will depend on your network and 

lib/replication_event.go

@@ -3,9 +3,8 @@ package lib
 import "github.com/fxamacker/cbor/v2"
 
 type ReplicationEvent[T any] struct {
-	FromNodeId  uint64
-	ChangeRowId int64
-	Payload     *T
+	FromNodeId uint64
+	Payload    *T
 }
 
 func (e *ReplicationEvent[T]) Marshal() ([]byte, error) {

lib/replicator.go

@@ -8,11 +8,12 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
+const maxReplcateRetries = 7
 const DefaultUrl = nats.DefaultURL
 const NodeNamePrefix = "marmot-node"
 
 var MaxLogEntries = int64(1024)
-var EntryReplicas = int(0)
+var EntryReplicas = 0
 var StreamNamePrefix = "marmot-changes"
 var SubjectPrefix = "marmot-change-log"
 
@@ -40,11 +41,10 @@ func NewReplicator(nodeID uint64, natsServer string, shards uint64) (*Replicator
 		}
 
 		streamCfg := makeShardConfig(shard, shards)
-		_, err = js.StreamInfo(streamCfg.Name)
-		if err != nil {
-			_, err = js.AddStream(streamCfg)
-		} else {
-			_, err = js.UpdateStream(streamCfg)
+		info, err := js.StreamInfo(streamCfg.Name)
+		if err == nats.ErrStreamNotFound {
+			log.Debug().Uint64("shard", shard).Msg("Creating stream")
+			info, err = js.AddStream(streamCfg)
 		}
 
 		if err != nil {
@@ -53,7 +53,14 @@ func NewReplicator(nodeID uint64, natsServer string, shards uint64) (*Replicator
 
 		log.Debug().
 			Uint64("shard", shard).
-			Msg("Created stream")
+			Str("name", info.Config.Name).
+			Int("replicas", info.Config.Replicas).
+			Str("leader", info.Cluster.Leader).
+			Msg("Stream ready...")
+
+		if err != nil {
+			return nil, err
+		}
 
 		streamMap[shard] = js
 	}
@@ -89,14 +96,14 @@ func (r *Replicator) Listen(shardID uint64, callback func(payload []byte) error)
 	sub, err := js.SubscribeSync(
 		subjectName(shardID),
 		nats.Durable(nodeName(r.nodeID)),
-		nats.ManualAck(),
 	)
 
 	if err != nil {
 		return err
 	}
 	defer sub.Unsubscribe()
 
+	replRetry := 0
 	for sub.IsValid() {
 		msg, err := sub.NextMsg(1 * time.Second)
 
@@ -111,9 +118,17 @@ func (r *Replicator) Listen(shardID uint64, callback func(payload []byte) error)
 		log.Debug().Str("sub", msg.Subject).Uint64("shard", shardID).Send()
 		err = callback(msg.Data)
 		if err != nil {
-			return err
+			if replRetry > maxReplcateRetries {
+				return err
+			}
+
+			log.Error().Err(err).Msg("Unable to process message retrying")
+			msg.Nak()
+			replRetry++
+			continue
 		}
 
+		replRetry = 0
 		err = msg.Ack()
 		if err != nil {
 			return err

marmot.go

@@ -3,6 +3,7 @@ package main
 import (
 	"flag"
 	"math/rand"
+	"time"
 
 	"github.com/maxpert/marmot/db"
 	"github.com/maxpert/marmot/lib"
@@ -67,20 +68,35 @@ func main() {
 
 	errChan := make(chan error)
 	for i := uint64(0); i < *shards; i++ {
-		go changeListener(streamDB, rep, i+1)
+		go changeListener(streamDB, rep, i+1, errChan)
 	}
 
-	err = <-errChan
-	if err != nil {
-		log.Panic().Err(err).Msg("Terminated listener")
+	cleanupInterval := 5 * time.Second
+	cleanupTicker := time.NewTicker(cleanupInterval)
+	defer cleanupTicker.Stop()
+
+	for {
+		select {
+		case err = <-errChan:
+			if err != nil {
+				log.Panic().Err(err).Msg("Terminated listener")
+			}
+		case t := <-cleanupTicker.C:
+			cnt, err := streamDB.CleanupChangeLogs(t.Add(-cleanupInterval))
+			if err != nil {
+				log.Warn().Err(err).Msg("Unable to cleanup change logs")
+			} else if cnt > 0 {
+				log.Debug().Int64("count", cnt).Msg("Cleaned up DB change logs")
+			}
+		}
 	}
 }
 
-func changeListener(streamDB *db.SqliteStreamDB, rep *lib.Replicator, shard uint64) {
+func changeListener(streamDB *db.SqliteStreamDB, rep *lib.Replicator, shard uint64, errChan chan error) {
 	log.Debug().Uint64("shard", shard).Msg("Listening stream")
 	err := rep.Listen(shard, onChangeEvent(streamDB))
 	if err != nil {
-		log.Panic().Err(err).Msg("Listener error")
+		errChan <- err
 	}
 }
 
@@ -89,24 +105,20 @@ func onChangeEvent(streamDB *db.SqliteStreamDB) func(data []byte) error {
 		ev := &lib.ReplicationEvent[db.ChangeLogEvent]{}
 		err := ev.Unmarshal(data)
 		if err != nil {
+			log.Error().Err(err).Send()
 			return err
 		}
 
-		ok, _ := streamDB.DeleteChangeLog(ev.Payload)
-		if ok {
-			return nil
-		}
-
+		_, _ = streamDB.DeleteChangeLog(ev.Payload)
 		return streamDB.Replicate(ev.Payload)
 	}
 }
 
 func onTableChanged(r *lib.Replicator, nodeID uint64, shards uint64) func(event *db.ChangeLogEvent) error {
 	return func(event *db.ChangeLogEvent) error {
 		ev := &lib.ReplicationEvent[db.ChangeLogEvent]{
-			FromNodeId:  nodeID,
-			ChangeRowId: event.Id,
-			Payload:     event,
+			FromNodeId: nodeID,
+			Payload:    event,
 		}
 
 		data, err := ev.Marshal()
@@ -119,7 +131,7 @@ func onTableChanged(r *lib.Replicator, nodeID uint64, shards uint64) func(event
 			return err
 		}
 
-		err = r.Publish(hash%shards, data)
+		err = r.Publish(hash, data)
 		if err != nil {
 			return err
 		}