mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-02 12:22:18 +00:00
Vladyslav SimonenkoandGitHub
453ef64920
## What changed? Adds TestPGXSimpleProtocol — a new entry in the persistence integration suite that runs `PostgreSQLSuite` under the `postgres12_pgx` plugin with `default_query_exec_mode=simple_protocol`. ## Why? Regression coverage for issues like [#9804](https://github.com/temporalio/temporal/issues/9804). With pgx ≤ v5.9.1, current_executions.state/status (proto-enum-typed int32 fields with a String() method) were text-encoded via fmt.Stringer and rejected by `Postgres` on simple/exec protocol, the path users land on behind PgBouncer in transaction pooling. pgx v5.9.2 fixed it upstream; this test makes sure we notice if pgx is ever downgraded or if a similar issue sneaks in. ## How did you test it? - [X] built - [X] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [X] added new functional test(s) Verified locally: passes on pgx v5.9.2; fails reproducibly on v5.9.1 with invalid input syntax for type integer: "Created" matching [#9804](https://github.com/temporalio/temporal/issues/9804). ## Potential risks Adds one additional pass through PostgreSQLSuite to the Integration test job. Job timeout is 15 min, so should be fine, but worth observing.
88 lines
2.3 KiB
Go
88 lines
2.3 KiB
Go
package persistencetest
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/server/common/persistence"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
)
|
|
|
|
type (
|
|
getQueueKeyParams struct {
|
|
QueueType persistence.QueueV2Type
|
|
Category tasks.Category
|
|
}
|
|
)
|
|
|
|
func WithQueueType(queueType persistence.QueueV2Type) func(p *getQueueKeyParams) {
|
|
return func(p *getQueueKeyParams) {
|
|
p.QueueType = queueType
|
|
}
|
|
}
|
|
|
|
func WithCategory(category tasks.Category) func(p *getQueueKeyParams) {
|
|
return func(p *getQueueKeyParams) {
|
|
p.Category = category
|
|
}
|
|
}
|
|
|
|
func GetQueueKey(t *testing.T, opts ...func(p *getQueueKeyParams)) persistence.QueueKey {
|
|
params := &getQueueKeyParams{
|
|
QueueType: persistence.QueueTypeHistoryNormal,
|
|
Category: tasks.CategoryTransfer,
|
|
}
|
|
for _, opt := range opts {
|
|
opt(params)
|
|
}
|
|
// Note that it is important to include the test name in the cluster name to ensure that the generated queue name is
|
|
// unique across tests. That way, we can run many queue tests without any risk of queue name collisions.
|
|
return persistence.QueueKey{
|
|
QueueType: params.QueueType,
|
|
Category: params.Category,
|
|
SourceCluster: "src-" + t.Name(),
|
|
TargetCluster: "tgt-" + t.Name(),
|
|
}
|
|
}
|
|
|
|
type EnqueueParams struct {
|
|
Data []byte
|
|
EncodingType int
|
|
}
|
|
|
|
func EnqueueMessage(
|
|
ctx context.Context,
|
|
queue persistence.QueueV2,
|
|
queueType persistence.QueueV2Type,
|
|
queueName string,
|
|
opts ...func(p *EnqueueParams),
|
|
) (*persistence.InternalEnqueueMessageResponse, error) {
|
|
params := EnqueueParams{
|
|
Data: []byte("1"),
|
|
EncodingType: int(enumspb.ENCODING_TYPE_JSON),
|
|
}
|
|
for _, opt := range opts {
|
|
opt(¶ms)
|
|
}
|
|
return queue.EnqueueMessage(ctx, &persistence.InternalEnqueueMessageRequest{
|
|
QueueType: queueType,
|
|
QueueName: queueName,
|
|
Blob: &commonpb.DataBlob{
|
|
EncodingType: enumspb.EncodingType(params.EncodingType),
|
|
Data: params.Data,
|
|
},
|
|
})
|
|
}
|
|
|
|
func EnqueueMessagesForDelete(t *testing.T, q persistence.QueueV2, queueName string, queueType persistence.QueueV2Type) {
|
|
for range 2 {
|
|
// We have to actually enqueue 2 messages. Otherwise, there won't be anything to actually delete,
|
|
// since we never delete the last message.
|
|
_, err := EnqueueMessage(context.Background(), q, queueType, queueName)
|
|
require.NoError(t, err)
|
|
}
|
|
}
|