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

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 @@
 	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 @@
 	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 @@
 	"github.com/sirupsen/logrus"
 
 	"github.com/bitleak/lmstfy/engine"
+	"github.com/bitleak/lmstfy/uuid"
 )
 
 const (
@@ -31,6 +32,8 @@
 	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 @@
 		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 @@
 	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 @@
 
 	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)