fix: change log out of order bug (#14)

* fix: change log out of order bug

* nobug: fix readme and add newline to cli

.gitignore

@@ -1,16 +1,5 @@
 *__debug_bin*
-.vscode
-.idea
-.fleet
-
-infra/
-.aws/
-dist/
-./dgate
-./k6
-
-# Local files #
-.dgate*/
+.dgate
 cov.out
 go.work.sum
 .env

README.md

@@ -45,12 +45,3 @@ DGate CLI is a command-line interface that can be used to interact with the DGat
 - Error Handler Module (`errorHandler`) - executed when an error occurs when sending a request to the upstream server. This module is used to modify the response before it is sent to the client.
 
 - Request Handler Module (`requestHandler`) - executed when a request is received from the client. This module is used to handle arbitrary requests, instead of using an upstream service.
-
-
-
-- Examples
-  - [x] ip hash load balancer
-  - [x] short url service
-  - [x] modify json request/response
-  - [x] send multiple upstream requests and combine them
-

cmd/dgate-cli/main.go

@@ -16,5 +16,7 @@ func main() {
 		os.Stderr.WriteString(err.Error())
 		os.Stderr.WriteString("\n")
 		os.Exit(1)
+		return
 	}
+	os.Stdout.WriteString("\n")
 }

internal/proxy/change_log.go

@@ -31,8 +31,8 @@ func (ps *ProxyState) processChangeLog(cl *spec.ChangeLog, reload, store bool) (
 		defer func() {
 			if err == nil {
 				if !ps.raftEnabled {
-					// dont store change logs
-					if err = ps.store.StoreChangeLog(cl); err != nil {
+					// renew the change log ID to avoid out-of-order processing
+					if err = ps.store.StoreChangeLog(cl.RenewID()); err != nil {
 						ps.logger.Error("Error storing change log, restarting state", zap.Error(err))
 						return
 					}
@@ -346,12 +346,20 @@ func (ps *ProxyState) restoreFromChangeLogs(directApply bool) error {
 	}
 	ps.logger.Info("restoring state change logs from storage", zap.Int("count", len(logs)))
 	// we might need to sort the change logs by timestamp
-	for _, cl := range logs {
+	for i, cl := range logs {
 		// skip documents as they are persisted in the store
 		if cl.Cmd.Resource() == spec.Documents {
 			continue
 		}
 		if err = ps.processChangeLog(cl, false, false); err != nil {
+			ps.logger.Error("error processing change log",
+				zap.Bool("skip", ps.debugMode),
+				zap.Error(err),
+				zap.Int("index", i),
+			)
+			if ps.debugMode {
+				continue
+			}
 			return err
 		} else {
 			ps.changeLogs = append(ps.changeLogs, cl)
@@ -445,29 +453,3 @@ START:
 	removeList = sliceutil.SliceUnique(removeList, func(cl *spec.ChangeLog) string { return cl.ID })
 	return removeList
 }
-
-// Function to check if there is a delete command between two logs with matching keys
-// func hasDeleteBetween(logs []*spec.ChangeLog, start, end *spec.ChangeLog) bool {
-// 	startIndex := -1
-// 	endIndex := -1
-
-// 	for i, log := range logs {
-// 		if log.ID == start.ID {
-// 			startIndex = i
-// 		}
-// 		if log.ID == end.ID {
-// 			endIndex = i
-// 		}
-// 	}
-
-// 	if startIndex == -1 || endIndex == -1 {
-// 		return false
-// 	}
-
-// 	for i := startIndex + 1; i < endIndex; i++ {
-// 		if logs[i].Cmd.IsDeleteCommand() {
-// 			return true
-// 		}
-// 	}
-// 	return false
-// }

internal/proxy/proxy_state.go

@@ -299,7 +299,7 @@ func (ps *ProxyState) ApplyChangeLog(log *spec.ChangeLog) error {
 		if err != nil {
 			return err
 		}
-		raftLog := raft.Log{Data: encodedCL}
+		raftLog := raft.Log{ Data: encodedCL }
 		now := time.Now()
 		future := r.ApplyLog(raftLog, time.Second*15)
 		err = future.Error()

pkg/spec/change_log.go

@@ -43,6 +43,14 @@ func NewChangeLog(item Named, namespace string, cmd Command) *ChangeLog {
 	}
 }
 
+func (cl *ChangeLog) RenewID() *ChangeLog {
+	changeLog := *cl
+	changeLog.ID = strconv.FormatInt(
+		time.Now().UnixNano(), 36,
+	)
+	return &changeLog
+}
+
 type Command string
 
 type Action string