Queue unexpected type conversion #11572

@wxiaoguang commented on GitHub (Aug 31, 2023):

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

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

@wxiaoguang commented on GitHub (Aug 31, 2023): It's an old behavior inherited from the old system. Because, the "queue" uses JSON to marshal the items.

GiteaMirror commented

@wxiaoguang commented on GitHub (Aug 31, 2023):

Such behavior came from #9363 and #5363

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

@wxiaoguang commented on GitHub (Aug 31, 2023): Such behavior came from #9363 and #5363 Would you like to change the behavior in the future, or keep it?

GiteaMirror commented

@KN4CK3R commented on GitHub (Aug 31, 2023):

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.

@KN4CK3R commented on GitHub (Aug 31, 2023): 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.

GiteaMirror commented

@wxiaoguang commented on GitHub (Aug 31, 2023):

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)

@wxiaoguang commented on GitHub (Aug 31, 2023): > 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)

GiteaMirror commented

@KN4CK3R commented on GitHub (Aug 31, 2023):

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

@KN4CK3R commented on GitHub (Aug 31, 2023): Does not work in that case because I don't push a simple int but a struct where a member is an interface.

GiteaMirror commented

@wxiaoguang commented on GitHub (Aug 31, 2023):

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

@wxiaoguang commented on GitHub (Aug 31, 2023): I guess you need to use `string`, or wrap your content into another encoding (eg: base64).

GiteaMirror commented

@wxiaoguang commented on GitHub (Aug 31, 2023):

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).

@wxiaoguang commented on GitHub (Aug 31, 2023): 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). ![image](https://github.com/go-gitea/gitea/assets/2114189/8dfede0c-af7a-415d-a973-393ea8c4ed35)

GiteaMirror commented

@KN4CK3R commented on GitHub (Aug 31, 2023):

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)

@KN4CK3R commented on GitHub (Aug 31, 2023): 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 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) ```

GiteaMirror commented

@wxiaoguang commented on GitHub (Aug 31, 2023):

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?

@wxiaoguang commented on GitHub (Aug 31, 2023): > 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`?

GiteaMirror commented

@KN4CK3R commented on GitHub (Aug 31, 2023):

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.

@KN4CK3R commented on GitHub (Aug 31, 2023): 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.

GiteaMirror commented

@wxiaoguang commented on GitHub (Aug 31, 2023):

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}}

@wxiaoguang commented on GitHub (Aug 31, 2023): Didn't see any problem, what's your code? ```go 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}} ```

GiteaMirror commented

@KN4CK3R commented on GitHub (Aug 31, 2023):

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
        }

@KN4CK3R commented on GitHub (Aug 31, 2023): I used the queue code from the first post. ```diff 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 } ```

GiteaMirror commented

@wxiaoguang commented on GitHub (Aug 31, 2023):

What's your queue item T ?

@wxiaoguang commented on GitHub (Aug 31, 2023): What's your queue item `T` ?

GiteaMirror commented

@KN4CK3R commented on GitHub (Aug 31, 2023):

any

@KN4CK3R commented on GitHub (Aug 31, 2023): any

GiteaMirror commented