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

#225)

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

To mitigate this issue, we introduce the version prefix for the job id
to identify different job payload formats. 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 the job body
directly if the length is 26. Otherwise, decode it in JSON format.

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 the new job payload format by upgrading the server first, and enabling
it on the client side since then.

engine/job.go

@@ -2,9 +2,7 @@ package engine
 
 import (
 	"encoding"
-	"encoding/binary"
 	"encoding/json"
-	"errors"
 
 	"github.com/bitleak/lmstfy/uuid"
 )
@@ -20,8 +18,6 @@ type Job interface {
 	ElapsedMS() int64
 	Attributes() map[string]string
 
-	encoding.BinaryMarshaler
-	encoding.BinaryUnmarshaler
 	encoding.TextMarshaler
 }
 
@@ -43,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,
@@ -110,70 +106,6 @@ func (j *jobImpl) Attributes() map[string]string {
 	return j.attributes
 }
 
-// Marshal into binary of the format:
-// {total len: 4 bytes}{ns len: 1 byte}{ns}{queue len: 1 byte}{queue}{id: 16 bytes}{ttl: 4 bytes}{tries: 2 byte}{job data}
-func (j *jobImpl) MarshalBinary() (data []byte, err error) {
-	nsLen := len(j.namespace)
-	qLen := len(j.queue)
-	bodyLen := len(j.body)
-	totalSize := 1 + nsLen + 1 + qLen + 16 + 4 + 2 + bodyLen
-	buf := make([]byte, totalSize+4)
-	binary.LittleEndian.PutUint32(buf, uint32(totalSize))
-
-	nsOffset := 4 + 1
-	qOffset := nsOffset + nsLen + 1
-	idOffset := qOffset + qLen
-	ttlOffset := idOffset + 16
-	triesOffset := ttlOffset + 4
-	jobOffset := triesOffset + 2
-
-	buf[4] = uint8(nsLen)
-	copy(buf[nsOffset:], j.namespace)
-	buf[qOffset-1] = uint8(qLen)
-	copy(buf[qOffset:], j.queue)
-	binID := uuid.UniqueIDToBinary(j.id)
-	copy(buf[idOffset:], binID[:]) // binary ID is 16 byte-long
-	binary.LittleEndian.PutUint32(buf[ttlOffset:], j.ttl)
-	binary.LittleEndian.PutUint16(buf[triesOffset:], j.tries)
-	copy(buf[jobOffset:], j.body)
-	return buf, nil
-}
-
-func (j *jobImpl) UnmarshalBinary(data []byte) error {
-	if len(data) <= 4 {
-		return errors.New("data too small")
-	}
-	totalSize := binary.LittleEndian.Uint32(data[0:])
-	if len(data) != int(totalSize)+4 {
-		return errors.New("corrupted data")
-	}
-
-	nsLen := int(data[4])
-	nsOffset := 4 + 1
-	j.namespace = string(data[nsOffset : nsOffset+nsLen])
-	qOffset := nsOffset + nsLen + 1
-	qLen := int(data[qOffset-1])
-	j.queue = string(data[qOffset : qOffset+qLen])
-	idOffset := qOffset + qLen
-	var binaryID [16]byte
-	copy(binaryID[:], data[idOffset:idOffset+16])
-	j.id = uuid.BinaryToUniqueID(binaryID)
-	ttlOffset := idOffset + 16
-	j.ttl = binary.LittleEndian.Uint32(data[ttlOffset:])
-	triesOffset := ttlOffset + 4
-	j.tries = binary.LittleEndian.Uint16(data[triesOffset:])
-	jobOffset := triesOffset + 2
-	j.body = make([]byte, len(data)-jobOffset)
-	copy(j.body, data[jobOffset:])
-
-	delay, err := uuid.ExtractDelaySecondFromUniqueID(j.id)
-	if err != nil {
-		return err
-	}
-	j.delay = delay
-	return nil
-}
-
 func (j *jobImpl) MarshalText() (text []byte, err error) {
 	var job struct {
 		Namespace string `json:"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)