Continue to implement callback processor

.gitignore

@@ -28,5 +28,5 @@ go.work.sum
 .env
 
 # Editor/IDE
-# .idea/
-# .vscode/
+.idea/
+.vscode/

api.yaml

@@ -223,7 +223,6 @@ components:
           type: string
           description: Unique identifier for the timer within the namespace (used for deduplication)
           example: "user-notification-123"
-          maxLength: 255
         executeAt:
           type: string
           format: date-time
@@ -234,7 +233,6 @@ components:
           format: uri
           description: HTTP URL to call when the timer executes, returning 200 with CallbackResponse means success, 4xx means invalid timer and no retry, otherwise will be retried.
           example: "https://api.example.com/webhooks/timer"
-          maxLength: 2048
         payload:
           type: object
           description: Custom JSON payload to include in the callback
@@ -258,7 +256,9 @@ components:
         nextExecuteAt:
           type: string
           format: date-time
-          description: ISO 8601 timestamp when the timer should execute next time. Only valid when ok is false.
+          description: |
+            ISO 8601 timestamp when the timer should execute next time. Only valid when ok is false.
+            The retry policy will be reset to the initial attempt(since executeAt is changed)
           example: "2024-12-20T15:30:00Z"
 
     UpdateTimerRequest:
@@ -278,14 +278,13 @@ components:
         executeAt:
           type: string
           format: date-time
-          description: New execution time for the timer
+          description: New execution time for the timer. If provided, the retry policy will be reset to the initial attempt.
           example: "2024-12-21T15:30:00Z"
         callbackUrl:
           type: string
           format: uri
           description: New callback URL, returning 200 with CallbackResponse means success, 4xx means invalid timer and no retry, otherwise will be retried.
           example: "https://api.example.com/webhooks/timer-updated"
-          maxLength: 2048
         payload:
           type: object
           description: New payload data
@@ -317,7 +316,7 @@ components:
         executeAt:
           type: string
           format: date-time
-          description: When the timer is scheduled to execute
+          description: When the timer is scheduled to execute as the first attempt.
           example: "2024-12-20T15:30:00Z"
         callbackUrl:
           type: string
@@ -349,7 +348,7 @@ components:
         executedAt:
           type: string
           format: date-time
-          description: When the timer was executed (if applicable)
+          description: When the timer was executed and the callback failed (if applicable)
           example: "2024-12-20T15:30:05Z"
 
     RetryPolicy:

docker/dev-all.yaml

@@ -1,5 +1,3 @@
-version: '3.8'
-
 services:
   # Cassandra Database
   cassandra:

server/TODO.md

@@ -0,0 +1,28 @@
+In late august, I accidently lost all the process that I have made in this project :( ....
+
+which I have worked really hard for a few weeks, but I forgot to push to Github.
+
+Now I have to rely on my poor memory for what I have done during that time. I am listing them here as TODOs by go over every file:
+
+1. API schema
+   1.1 RetryPolicy need to change to use integers instead of strings
+   1.2 RetryPolicy in DbTimer need to use the actual type
+   1.3 Use go generator instead of go-gin-server
+
+
+2. DB
+   2.1 remove RangeDeleteWithBatchInsertTxn
+   2.2 implment shardId in DB layer
+   2.3 fix behavior of RangeGetTimersRequest to be exclusive + inclusive
+   2.4 Implement UpdadateTimerByUUID API for updating the next executeAt for nextExecuteAt and backoff retry case
+   2.5 Add a flag to indicate the timer callback has failed
+
+
+
+3. Engine
+   3.1 implement callback processor
+   3.2 add local backoff retry
+   3.3. implement timer queue
+
+4. Membership
+   4.1 reuse what have done in xcherry

server/databases/db_unified_interfaces.go

@@ -24,67 +24,69 @@ type (
 			metadata ShardMetadata,
 		) (err *DbError)
 
-		CreateTimer(
+		GetTimer(
 			ctx context.Context,
-			shardId int, shardVersion int64, namespace string,
-			timer *DbTimer,
-		) (err *DbError)
+			shardId int, namespace string, timerId string,
+		) (timer *DbTimer, err *DbError)
 
 		CreateTimerNoLock(
 			ctx context.Context,
 			shardId int, namespace string,
 			timer *DbTimer,
 		) (err *DbError)
 
+		UpdateTimerNoLock(
+			ctx context.Context,
+			shardId int, namespace string,
+			request *UpdateDbTimerRequest,
+		) (err *DbError)
+
+		DeleteTimerNoLock(
+			ctx context.Context,
+			shardId int, namespace string, timerId string,
+		) *DbError
+
 		RangeGetTimers(
 			ctx context.Context,
 			shardId int,
 			request *RangeGetTimersRequest,
 		) (*RangeGetTimersResponse, *DbError)
 
 		// RangeDeleteWithBatchInsertTxn is a transaction that deletes timers in a range and inserts new timers
+		// TODO: remove this
 		RangeDeleteWithBatchInsertTxn(
 			ctx context.Context,
 			shardId int, shardVersion int64,
 			request *RangeDeleteTimersRequest,
 			TimersToInsert []*DbTimer,
 		) (*RangeDeleteTimersResponse, *DbError)
 
-		// RangeDeleteWithLimit is a non-transactional operation that deletes timers in a range 
+		// RangeDeleteWithLimit is a non-transactional operation that deletes timers in a range
 		RangeDeleteWithLimit(
 			ctx context.Context,
 			shardId int,
 			request *RangeDeleteTimersRequest,
 			limit int, // Note that some distributed databases like Cassandra/MongoDB/DynamoDB don't support multiple range queries with LIMIT, so it may be ignored
 		) (*RangeDeleteTimersResponse, *DbError)
 
+		// TODO: only used in speical case (time skew)
 		UpdateTimer(
 			ctx context.Context,
 			shardId int, shardVersion int64, namespace string,
 			request *UpdateDbTimerRequest,
 		) (err *DbError)
 
-
-		GetTimer(
-			ctx context.Context,
-			shardId int, namespace string, timerId string,
-		) (timer *DbTimer, err *DbError)
-
+		// TODO: only used in speical case (time skew)
 		DeleteTimer(
 			ctx context.Context,
 			shardId int, shardVersion int64, namespace string, timerId string,
 		) *DbError
 
-		UpdateTimerNoLock(
+		// TODO: only used in speical case (time skew)
+		CreateTimer(
 			ctx context.Context,
-			shardId int, namespace string,
-			request *UpdateDbTimerRequest,
+			shardId int, shardVersion int64, namespace string,
+			timer *DbTimer,
 		) (err *DbError)
-
-
-		DeleteTimerNoLock(
-			ctx context.Context,
-			shardId int, namespace string, timerId string,
-		) *DbError
 	}
 )

server/databases/db_unified_models.go

@@ -45,6 +45,7 @@ type (
 
 	// DbTimer is the timer model stored in DB
 	DbTimer struct {
+		ShardId      int // TODO: implment this in DB layer
 
 		// Unique identifier for the timer
 		Id string