Queue unexpected type conversion

[KN4CK3R]

Description

q := queue.CreateSimpleQueue(
	graceful.GetManager().ShutdownContext(),
	"interface",
	func(data ...any) []any {
		for _, d := range data {
			fmt.Printf("<- %#v (%T)\n", d, d)
		}
		return nil
	},
)

go graceful.GetManager().RunWithCancel(q)

push := func(val any) {
	fmt.Printf("-> %#v (%T)\n", val, val)
	q.Push(val)
}

push(int64(1))
push(1)

Expected:

-> 1 (int64)
<- 1 (int64)
-> 1 (int)
<- 1 (int)

Actual:

-> 1 (int64)
<- 1 (float64)
-> 1 (int)
<- 1 (float64)

Somewhere there is an unexpected type conversion in the code.

@wxiaoguang You touched the queues last, it's yours 😄

[wxiaoguang]

It's an old behavior inherited from the old system.

Because, the "queue" uses JSON to marshal the items.

[wxiaoguang]

Such behavior came from #9363 and #5363

Would you like to change the behavior in the future, or keep it?

[KN4CK3R]

I thought so but only "expected" that behaviour when using external queues like redis. I found it yesterday when writing tests for my audit PR. The check assert.Equal(t, case.Scope.PrimaryKey, expected.Scope.PrimaryKey) failed because of the different types.
I think we should use a different marshal method like https://pkg.go.dev/encoding/gob

Why do we need marshalling at all in the normal flow? I would expect that it is only used when writing the queue to disk or similar.

[wxiaoguang]

The check assert.Equal(t, case.Scope.PrimaryKey, expected.Scope.PrimaryKey) failed because of the different types.

At the moment, it is not a serous problem, because you could use typed handler: func(data ...int64) []int64 {, then there won't be any problem (do not use any)

Why do we need marshalling at all in the normal flow? I would expect that it is only used when writing the queue to disk or similar.

Yes, for persistence (disk/redis) and consistency (the same behavior for all queues)

[KN4CK3R]

Does not work in that case because I don't push a simple int but a struct where a member is an interface.

[wxiaoguang]

I guess you need to use string, or wrap your content into another encoding (eg: base64).

[wxiaoguang]

ps: I guess I know why you are surprised now ....

Your audit PR was before my Rewrite queue #24505

The background is: Gitea runs tests with "immediate" queue.

Before the queue refactoring, the "immediate" queue behave quite differently from the standard queues. It just calls the handler directly (no marshal/unmarshal)

After the refactoring, I have done my best to make "immediate" queue behave as much same as the standard queus (eg: do q.unmarshal(q.marshal(data)) to call the handler)

So, the "JSON unmarshal any" problem in code gets caught by the new queue code (luckily).

[KN4CK3R]

The tests are new, so I'm surprised unrelated to your rewrite. Tested MsgPack and that seems to be a working replacement (but no drop in replacement because of existing persistent stores):

diff --git a/modules/queue/workerqueue.go b/modules/queue/workerqueue.go
index b28fd8802..c3404a619 100644
--- a/modules/queue/workerqueue.go
+++ b/modules/queue/workerqueue.go
@@ -10,10 +10,11 @@ import (
        "sync/atomic"
        "time"
 
-       "code.gitea.io/gitea/modules/json"
        "code.gitea.io/gitea/modules/log"
        "code.gitea.io/gitea/modules/process"
        "code.gitea.io/gitea/modules/setting"
+
+       "github.com/vmihailenco/msgpack/v5"
 )
 
 // WorkerPoolQueue is a queue that uses a pool of workers to process items
@@ -137,16 +138,16 @@ func (q *WorkerPoolQueue[T]) RemoveAllItems(ctx context.Context) error {
 }
 
 func (q *WorkerPoolQueue[T]) marshal(data T) []byte {
-       bs, err := json.Marshal(data)
+       bs, err := msgpack.Marshal(data)
        if err != nil {
                log.Error("Failed to marshal item for queue %q: %v", q.GetName(), err)
                return nil
        }
        return bs
 }
 
 func (q *WorkerPoolQueue[T]) unmarshal(data []byte) (t T, ok bool) {
-       if err := json.Unmarshal(data, &t); err != nil {
+       if err := msgpack.Unmarshal(data, &t); err != nil {
                log.Error("Failed to unmarshal item from queue %q: %v", q.GetName(), err)
                return t, false
        }

-> 1 (int64)
<- 1 (int64)
-> 1 (int32)
<- 1 (int32)
-> "1" (string)
<- "1" (string)

[wxiaoguang]

Tested MsgPack and that seems to be a working replacement (but no drop in replacement because of existing persistent stores):

It seems worth to try.

ps: why not encoding/gob?

[KN4CK3R]

Tested that but it's not flexible enough because you need to register (some) custom types:

Failed to unmarshal item from queue "interface": gob: local interface type *interface {} can only be decoded from remote interface type; received concrete type int

I stopped testing at that point but it may be a simple error.

[wxiaoguang]

Didn't see any problem, what's your code?

package main

import (
	"bytes"
	"encoding/gob"
	"fmt"
	"log"
)

type Sub struct {
	V any
}

type Msg struct {
	V1, V2 any
}

func main() {
	gob.Register(Sub{})

	buf := new(bytes.Buffer)
	enc := gob.NewEncoder(buf)
	dec := gob.NewDecoder(buf)

	err := enc.Encode(Msg{1, Sub{2}})
	if err != nil {
		log.Fatal("encode error:", err)
	}

	var m Msg
	err = dec.Decode(&m)
	if err != nil {
		log.Fatal("decode error 1:", err)
	}
	fmt.Printf("%#v\n", m)
}

main.Msg{V1:1, V2:main.Sub{V:2}}

[KN4CK3R]

I used the queue code from the first post.

 func (q *WorkerPoolQueue[T]) marshal(data T) []byte {
-       bs, err := json.Marshal(data)
-       if err != nil {
+       var buf bytes.Buffer
+       if err := gob.NewEncoder(&buf).Encode(data); err != nil {
                log.Error("Failed to marshal item for queue %q: %v", q.GetName(), err)
                return nil
        }
-       return bs
+       return buf.Bytes()
 }
 
 func (q *WorkerPoolQueue[T]) unmarshal(data []byte) (t T, ok bool) {
-       if err := json.Unmarshal(data, &t); err != nil {
+       if err := gob.NewDecoder(bytes.NewReader(data)).Decode(&t); err != nil {
                log.Error("Failed to unmarshal item from queue %q: %v", q.GetName(), err)
                return t, false
        }