Add support of parsing job body format according to job id version

Currently, we store the job payload in Redis without any encoding,
so it's possible to extend more fields for job, like attributes etc.

To mitigate this issue, we introduce the version prefix for the job id
to identify different job payload formats. And use the length to tell
if it's a legacy id or not since the ulid's id is a fixed length
string(26 chars). That said, we will return the value as job body
directly if the length is 26. Otherwise, decode it as JSON format.

And to avoid introducing breaking changes during the upgrade stage,
we add a new HTTP header: Enable-Job-Version to enable this feature.
The new job format would be enabled only if the header `Enable-Job-Version: yes`
was explicitly passed in the publish request. So that we can smoothly
support new job payload format by upgrading the server first, and enable
it in client side since then.

engine/job.go

@@ -39,7 +39,7 @@ type jobImpl struct {
 // a tombstone record in that AOF.
 func NewJob(namespace, queue string, body []byte, ttl, delay uint32, tries uint16, jobID string) Job {
 	if jobID == "" {
-		jobID = uuid.GenUniqueJobIDWithDelay(delay)
+		jobID = uuid.GenJobIDWithVersion(0, delay)
 	}
 	return &jobImpl{
 		namespace: namespace,

engine/redis/pool.go

@@ -1,14 +1,20 @@
 package redis
 
 import (
+	"encoding/json"
 	"errors"
 	"time"
 
 	go_redis "github.com/go-redis/redis/v8"
 
 	"github.com/bitleak/lmstfy/engine"
+	"github.com/bitleak/lmstfy/uuid"
 )
 
+type JobPayload struct {
+	Body []byte `json:"body"`
+}
+
 // Pool stores all the jobs' data. this is a global singleton per engine
 // note: this `Pool` is NOT the same terminology as the EnginePool
 type Pool struct {
@@ -33,14 +39,24 @@ func PoolJobKeyPrefix(namespace, queue string) string {
 	return join(PoolPrefix, namespace, queue)
 }
 
-func (p *Pool) Add(j engine.Job) error {
-	body := j.Body()
+func (p *Pool) Add(j engine.Job) (err error) {
 	metrics.poolAddJobs.WithLabelValues(p.redis.Name).Inc()
+
+	// For the version 0(legacy) jobID, the payload is the body directly,
+	// for the version 1 jobID, the payload is a JSON string contains the body.
+	payload := j.Body()
+	if uuid.ExtractJobIDVersion(j.ID()) != 0 {
+		payload, err = json.Marshal(JobPayload{Body: j.Body()})
+		if err != nil {
+			return err
+		}
+	}
+
 	// SetNX return OK(true) if key didn't exist before.
-	ok, err := p.redis.Conn.SetNX(dummyCtx, PoolJobKey(j), body, time.Duration(j.TTL())*time.Second).Result()
+	ok, err := p.redis.Conn.SetNX(dummyCtx, PoolJobKey(j), payload, time.Duration(j.TTL())*time.Second).Result()
 	if err != nil {
 		// Just retry once.
-		ok, err = p.redis.Conn.SetNX(dummyCtx, PoolJobKey(j), body, time.Duration(j.TTL())*time.Second).Result()
+		ok, err = p.redis.Conn.SetNX(dummyCtx, PoolJobKey(j), payload, time.Duration(j.TTL())*time.Second).Result()
 	}
 	if err != nil {
 		return err
@@ -57,24 +73,35 @@ func (p *Pool) Get(namespace, queue, jobID string) (body []byte, ttlSecond uint3
 	getCmd := pipeline.Get(dummyCtx, jobKey)
 	ttlCmd := pipeline.TTL(dummyCtx, jobKey)
 	_, err = pipeline.Exec(dummyCtx)
-	switch err {
-	case nil:
-		val := getCmd.Val()
-		ttl := int64(ttlCmd.Val().Seconds())
-		if ttl < 0 {
-			// Use `0` to identify indefinite TTL, NOTE: in redis ttl=0 is possible when
-			// the key is not recycled fast enough. but here is okay we use `0` to identify
-			// indefinite TTL, because we issue GET cmd before TTL cmd, so the ttl must be > 0,
-			// OR GET cmd would fail.
-			ttl = 0
+	if err != nil {
+		if errors.Is(err, go_redis.Nil) {
+			return nil, 0, engine.ErrNotFound
 		}
-		metrics.poolGetJobs.WithLabelValues(p.redis.Name).Inc()
-		return []byte(val), uint32(ttl), nil
-	case go_redis.Nil:
-		return nil, 0, engine.ErrNotFound
-	default:
 		return nil, 0, err
 	}
+
+	val := []byte(getCmd.Val())
+	ttl := int64(ttlCmd.Val().Seconds())
+	if ttl < 0 {
+		// Use `0` to identify indefinite TTL, NOTE: in redis ttl=0 is possible when
+		// the key is not recycled fast enough. but here is okay we use `0` to identify
+		// indefinite TTL, because we issue GET cmd before TTL cmd, so the ttl must be > 0,
+		// OR GET cmd would fail.
+		ttl = 0
+	}
+	metrics.poolGetJobs.WithLabelValues(p.redis.Name).Inc()
+	if uuid.ExtractJobIDVersion(jobID) == 0 {
+		// For the version 0(legacy) jobID, the val only contains the body,
+		// so we need to return the val as body directly.
+		return val, uint32(ttl), nil
+	}
+	// For the version 1 jobID, the value is encoded as a JSON string,
+	// need to unmarshal it before return.
+	var payload JobPayload
+	if err := json.Unmarshal(val, &payload); err != nil {
+		return nil, 0, err
+	}
+	return payload.Body, uint32(ttl), nil
 }
 
 func (p *Pool) Delete(namespace, queue, jobID string) error {

engine/redis/pool_test.go

@@ -6,8 +6,10 @@ import (
 	"time"
 
 	go_redis "github.com/go-redis/redis/v8"
+	"github.com/stretchr/testify/require"
 
 	"github.com/bitleak/lmstfy/engine"
+	"github.com/bitleak/lmstfy/uuid"
 )
 
 func TestPool_Add(t *testing.T) {
@@ -55,3 +57,20 @@ func TestPool_Get(t *testing.T) {
 		t.Fatalf("Expected TTL is around 50 seconds")
 	}
 }
+
+func TestPool_GetCompatibility(t *testing.T) {
+	p := NewPool(R)
+
+	t.Run("test job with different versions should get correct body", func(t *testing.T) {
+		for i := 0; i <= uuid.JobIDV1; i++ {
+			jobID := uuid.GenJobIDWithVersion(i, 123)
+			job := engine.NewJob("ns-pool", "q5", []byte("hello msg 5"), 50, 0, 1, jobID)
+			p.Add(job)
+			body, ttl, err := p.Get(job.Namespace(), job.Queue(), job.ID())
+			require.NoError(t, err)
+			require.Equal(t, []byte("hello msg 5"), body)
+			require.InDelta(t, 50, ttl, 5)
+			require.Equal(t, i, uuid.ExtractJobIDVersion(job.ID()))
+		}
+	})
+}

server/handlers/queue.go

@@ -10,6 +10,7 @@ import (
 	"github.com/sirupsen/logrus"
 
 	"github.com/bitleak/lmstfy/engine"
+	"github.com/bitleak/lmstfy/uuid"
 )
 
 const (
@@ -31,6 +32,8 @@ func Publish(c *gin.Context) {
 	queue := c.Param("queue")
 	jobID := c.Param("job_id")
 
+	enabledJobVersion := strings.ToUpper(c.GetHeader("Enable-Job-Version")) == "YES"
+
 	if jobID != "" {
 		// delete job whatever other publish parameters
 		if err := e.Delete(namespace, queue, jobID); err != nil {
@@ -85,7 +88,14 @@ func Publish(c *gin.Context) {
 		c.JSON(http.StatusRequestEntityTooLarge, gin.H{"error": "body too large"})
 		return
 	}
-	job := engine.NewJob(namespace, queue, body, uint32(ttlSecond), uint32(delaySecond), uint16(tries), "")
+
+	if enabledJobVersion {
+		jobID = uuid.GenJobIDWithVersion(uuid.JobIDV1, uint32(delaySecond))
+	} else {
+		// use the legacy jobID if the version is not enabled
+		jobID = uuid.GenJobIDWithVersion(0, uint32(delaySecond))
+	}
+	job := engine.NewJob(namespace, queue, body, uint32(ttlSecond), uint32(delaySecond), uint16(tries), jobID)
 	jobID, err = e.Publish(job)
 	if err != nil {
 		logger.WithFields(logrus.Fields{
@@ -122,6 +132,8 @@ func PublishBulk(c *gin.Context) {
 	namespace := c.Param("namespace")
 	queue := c.Param("queue")
 
+	enabledJobVersion := strings.ToUpper(c.GetHeader("Enable-Job-Version")) == "YES"
+
 	delaySecondStr := c.DefaultQuery("delay", DefaultDelay)
 	delaySecond, err := strconv.ParseUint(delaySecondStr, 10, 32)
 	if err != nil {
@@ -180,7 +192,14 @@ func PublishBulk(c *gin.Context) {
 
 	jobIDs := make([]string, 0)
 	for _, job := range jobs {
-		j := engine.NewJob(namespace, queue, job, uint32(ttlSecond), uint32(delaySecond), uint16(tries), "")
+		var jobID string
+		if enabledJobVersion {
+			jobID = uuid.GenJobIDWithVersion(uuid.JobIDV1, uint32(delaySecond))
+		} else {
+			// use the legacy jobID if the version is not enabled
+			jobID = uuid.GenJobIDWithVersion(0, uint32(delaySecond))
+		}
+		j := engine.NewJob(namespace, queue, job, uint32(ttlSecond), uint32(delaySecond), uint16(tries), jobID)
 		jobID, err := e.Publish(j)
 		if err != nil {
 			logger.WithFields(logrus.Fields{

server/handlers/queue_test.go

@@ -12,9 +12,11 @@ import (
 	"time"
 
 	"github.com/magiconair/properties/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/bitleak/lmstfy/engine"
 	"github.com/bitleak/lmstfy/server/handlers"
+	"github.com/bitleak/lmstfy/uuid"
 )
 
 func TestPublish(t *testing.T) {
@@ -543,6 +545,41 @@ func TestPublishBulk(t *testing.T) {
 	}
 }
 
+func TestPublish_WithJobVersion(t *testing.T) {
+	for _, enable := range []string{"YES", "NO"} {
+		query := url.Values{}
+		query.Add("delay", "0")
+		query.Add("ttl", "10")
+		query.Add("tries", "1")
+		targetUrl := fmt.Sprintf("http://localhost/api/ns/q18?%s", query.Encode())
+		body := strings.NewReader("hello job version")
+		req, err := http.NewRequest("PUT", targetUrl, body)
+		req.Header.Add("Enable-Job-Version", enable)
+		require.NoError(t, err, "Failed to create request")
+
+		c, e, resp := ginTest(req)
+		e.Use(handlers.ValidateParams, handlers.SetupQueueEngine)
+		e.PUT("/api/:namespace/:queue", handlers.Publish)
+		e.HandleContext(c)
+
+		require.Equal(t, http.StatusCreated, resp.Code, "Failed to publish")
+		var payload struct {
+			JobID string `json:"job_id"`
+		}
+		require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &payload))
+		expectedVersion := 0
+		if enable == "YES" {
+			expectedVersion = uuid.JobIDV1
+		}
+		require.Equal(t, expectedVersion, uuid.ExtractJobIDVersion(payload.JobID))
+
+		// Consume should also return the correct version and job body
+		bytes, jobID := consumeTestJob("ns", "q18", 10, 3)
+		require.Equal(t, expectedVersion, uuid.ExtractJobIDVersion(jobID))
+		require.Equal(t, "hello job version", string(bytes))
+	}
+}
+
 func publishTestJob(ns, q string, delay, ttl uint32) (body []byte, jobID string) {
 	e := engine.GetEngine("")
 	body = make([]byte, 10)

uuid/uuid.go

@@ -3,14 +3,15 @@ package uuid
 import (
 	"encoding/binary"
 	"errors"
+	"fmt"
 	"math/rand"
 	"sync"
 	"time"
 
 	"github.com/oklog/ulid"
 )
 
-const JobIDVersion = "1"
+const JobIDV1 = 1
 
 // Use pool to avoid concurrent access for rand.Source
 var entropyPool = sync.Pool{
@@ -28,21 +29,31 @@ func GenUniqueID() string {
 	return id.String()
 }
 
+// GenJobIDWithVersion generates a job ID with version prefix and delaySecond.
+// For the legacy version 0 job ID, the version prefix is not included,
+// we use the version prefix to distinguish different job payload format.
+//
 // Use the last four bytes of the 16-byte's ULID to store the delaySecond.
 // The last fours bytes was some random value in ULID, so changing that value won't
 // affect anything except randomness.
-func GenUniqueJobIDWithDelay(delaySecond uint32) string {
+func GenJobIDWithVersion(version int, delaySecond uint32) string {
 	entropy := entropyPool.Get().(*rand.Rand)
 	defer entropyPool.Put(entropy)
 	id := ulid.MustNew(ulid.Now(), entropy)
 	// Encode the delayHour in littleEndian and store at the last four bytes
 	binary.LittleEndian.PutUint32(id[len(id)-4:], delaySecond)
-	// Add a version prefix to identify the jobID format
-	return JobIDVersion + id.String()
+	// legacy version is 0, it doesn't include version prefix in the id
+	if version == 0 {
+		return id.String()
+	}
+	if version < 0 || version > 9 {
+		version = JobIDV1
+	}
+	return fmt.Sprintf("%d%s", version, id.String())
 }
 
 func ElapsedMilliSecondFromUniqueID(s string) (int64, error) {
-	s, _ = ExtractJobID(s)
+	s, _ = extractJobID(s)
 	id, err := ulid.Parse(s)
 	if err != nil {
 		return 0, err
@@ -57,17 +68,24 @@ func ElapsedMilliSecondFromUniqueID(s string) (int64, error) {
 }
 
 func ExtractDelaySecondFromUniqueID(s string) (uint32, error) {
-	s, _ = ExtractJobID(s)
+	s, _ = extractJobID(s)
 	id, err := ulid.Parse(s)
 	if err != nil {
 		return 0, err
 	}
 	return binary.LittleEndian.Uint32(id[len(id)-4:]), nil
 }
 
-func ExtractJobID(s string) (string, int) {
-	if len(s) == ulid.EncodedSize+1 {
-		return s[1:], int(s[0] - '0')
+func extractJobID(s string) (string, int) {
+	if len(s) <= ulid.EncodedSize {
+		return s, 0
+	}
+	return s[1:], int(s[0] - '0')
+}
+
+func ExtractJobIDVersion(s string) int {
+	if len(s) == ulid.EncodedSize {
+		return 0
 	}
-	return s, 0
+	return int(s[0] - '0')
 }

uuid/uuid_test.go

@@ -9,9 +9,9 @@ import (
 )
 
 func TestJobID(t *testing.T) {
-	jobID := GenUniqueJobIDWithDelay(10)
+	jobID := GenJobIDWithVersion(JobIDV1, 10)
 
-	id, version := ExtractJobID(jobID)
+	id, version := extractJobID(jobID)
 	require.Equal(t, 1, version)
 	require.Equal(t, ulid.EncodedSize, len(id))