diff --git a/agfs-server/go.mod b/agfs-server/go.mod index faa0900..ac7447f 100644 --- a/agfs-server/go.mod +++ b/agfs-server/go.mod @@ -36,8 +36,14 @@ require ( github.com/aws/aws-sdk-go-v2/service/sts v1.38.6 // indirect github.com/aws/smithy-go v1.23.0 // indirect github.com/hashicorp/golang-lru v0.5.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/pgx/v5 v5.9.1 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/klauspost/cpuid/v2 v2.0.9 // indirect + golang.org/x/sync v0.17.0 // indirect golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect + golang.org/x/text v0.29.0 // indirect ) replace github.com/c4pt0r/agfs/agfs-sdk/go => ../agfs-sdk/go diff --git a/agfs-server/go.sum b/agfs-server/go.sum index 7b6508b..9be3a7d 100644 --- a/agfs-server/go.sum +++ b/agfs-server/go.sum @@ -51,6 +51,14 @@ github.com/hashicorp/go-uuid v1.0.0 h1:RS8zrF7PhGwyNPOtxSClXXj9HA8feRnJzgnI1RJCS github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.9.1 h1:uwrxJXBnx76nyISkhr33kQLlUqjv7et7b9FjCen/tdc= +github.com/jackc/pgx/v5 v5.9.1/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4= github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/mattn/go-sqlite3 v1.14.32 h1:JD12Ag3oLy1zQA+BNn74xRgaBbdhbNIDYvQUEuuErjs= @@ -60,6 +68,7 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/tetratelabs/wazero v1.9.0 h1:IcZ56OuxrtaEz8UYNRHBrUa9bYeX9oVY93KspZZBf/I= @@ -68,8 +77,12 @@ github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ= github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0= github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0= github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ= golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= +golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/agfs-server/pkg/plugins/queuefs/README.md b/agfs-server/pkg/plugins/queuefs/README.md index ab5d01f..298a286 100644 --- a/agfs-server/pkg/plugins/queuefs/README.md +++ b/agfs-server/pkg/plugins/queuefs/README.md @@ -69,6 +69,84 @@ agfs:/> cat /queuefs/dequeue {"id":"...","data":"task-123","timestamp":"..."} ``` +## Running Tests + +From `agfs/agfs-server`, run the full `queuefs` test suite with: + +```bash +go test ./pkg/plugins/queuefs/... +``` + +This covers the default in-memory tests plus the local SQLite-backed regression tests. +TiDB and PostgreSQL integration tests are gated separately and only run when explicitly enabled. + +## Running SQLite Tests + +`queuefs` includes regression tests that exercise the real SQLite-backed implementation. + +To run only the SQLite-specific regression tests and bypass the Go test cache: + +```bash +go test ./pkg/plugins/queuefs -run 'TestQueueFSSQLite' -count=1 -v +``` + +These tests create temporary `.db` files with `t.TempDir()` and initialize the plugin with: + +```go +map[string]interface{}{ + "backend": "sqlite", + "db_path": dbPath, +} +``` + +The current SQLite regression coverage lives in `agfs-server/pkg/plugins/queuefs/sqlite_backend_test.go`. + +## Running TiDB Tests + +`queuefs` also includes gated integration tests for the TiDB backend. + +For a local playground started with: + +```bash +tiup playground v8.5.5 --tiflash 0 --without-monitor +``` + +From `agfs/agfs-server`, run: + +```bash +TIDB_TEST=1 \ +TIDB_TEST_DSN='root@tcp(127.0.0.1:4000)/queuedb?charset=utf8mb4&parseTime=True' \ +go test ./pkg/plugins/queuefs -run 'TestQueueFSTiDB' -count=1 -v +``` + +Notes: + +- The TiDB tests are skipped unless `TIDB_TEST=1` is set. +- `TIDB_TEST_DSN` is required; the tests replace the database name in that DSN with a fresh per-test database. +- The tests create a fresh database name for each run, so they do not reuse prior queue tables. +- The current TiDB regression coverage lives in `agfs-server/pkg/plugins/queuefs/tidb_backend_test.go`. + +## Running PostgreSQL Tests + +`queuefs` also includes gated integration tests for the PostgreSQL backend. + +The default PostgreSQL service is available at `127.0.0.1:5432`. + +From `agfs/agfs-server`, run: + +```bash +PG_TEST=1 \ +PG_TEST_DSN="postgresql://${USER}@127.0.0.1:5432/postgres?sslmode=disable" \ +go test ./pkg/plugins/queuefs -run 'TestQueueFSPGSQL' -count=1 -v +``` + +Notes: + +- The PostgreSQL tests are skipped unless `PG_TEST=1` is set. +- `PG_TEST_DSN` is required; the tests replace the database name in that DSN with a fresh per-test database. +- The tests create a fresh database name for each run, so they do not reuse prior queue tables. +- The current PostgreSQL regression coverage lives in `agfs-server/pkg/plugins/queuefs/pgsql_backend_test.go`. + ## License Apache License 2.0 diff --git a/agfs-server/pkg/plugins/queuefs/backend.go b/agfs-server/pkg/plugins/queuefs/backend.go index f2ccde9..6b2196f 100644 --- a/agfs-server/pkg/plugins/queuefs/backend.go +++ b/agfs-server/pkg/plugins/queuefs/backend.go @@ -218,8 +218,8 @@ func (b *MemoryBackend) QueueExists(queueName string) (bool, error) { return exists, nil } -// TiDBBackend implements QueueBackend using TiDB database -type TiDBBackend struct { +// SQLQueueBackend implements QueueBackend using a SQL database. +type SQLQueueBackend struct { db *sql.DB backend DBBackend backendType string @@ -227,26 +227,52 @@ type TiDBBackend struct { cacheMu sync.RWMutex // protects tableCache } -func NewTiDBBackend() *TiDBBackend { - return &TiDBBackend{ - tableCache: make(map[string]string), +func newSQLQueueBackend(backendType string, backend DBBackend) *SQLQueueBackend { + return &SQLQueueBackend{ + backendType: backendType, + backend: backend, + tableCache: make(map[string]string), } } -func (b *TiDBBackend) Initialize(config map[string]interface{}) error { - // Store backend type from config - backendType := "memory" // default - if val, ok := config["backend"]; ok { - if strVal, ok := val.(string); ok { - backendType = strVal +func NewSQLQueueBackend() *SQLQueueBackend { + return newSQLQueueBackend("", nil) +} + +func NewSQLiteQueueBackend() *SQLQueueBackend { + return newSQLQueueBackend("sqlite", NewSQLiteDBBackend()) +} + +func NewTiDBQueueBackend() *SQLQueueBackend { + return newSQLQueueBackend("tidb", NewTiDBDBBackend()) +} + +func NewPostgresQueueBackend() *SQLQueueBackend { + return newSQLQueueBackend("pgsql", NewPostgreSQLDBBackend()) +} + +func (b *SQLQueueBackend) Initialize(config map[string]interface{}) error { + backendType := b.backendType + if backendType == "" { + if val, ok := config["backend"]; ok { + if strVal, ok := val.(string); ok { + backendType = strVal + } } } + if backendType == "" || backendType == "memory" { + return fmt.Errorf("SQLQueueBackend requires a SQL backend (sqlite, tidb/mysql, pgsql/postgres); use MemoryBackend for in-memory mode") + } b.backendType = backendType // Create database backend - backend, err := CreateBackend(config) - if err != nil { - return fmt.Errorf("failed to create backend: %w", err) + backend := b.backend + if backend == nil { + var err error + backend, err = CreateBackend(config) + if err != nil { + return fmt.Errorf("failed to create backend: %w", err) + } } b.backend = backend @@ -268,20 +294,37 @@ func (b *TiDBBackend) Initialize(config map[string]interface{}) error { return nil } -func (b *TiDBBackend) Close() error { +func (b *SQLQueueBackend) Close() error { if b.db != nil { return b.db.Close() } return nil } -func (b *TiDBBackend) GetType() string { +func (b *SQLQueueBackend) GetType() string { return b.backendType } +func (b *SQLQueueBackend) rebind(query string) string { + if b.backend == nil { + return query + } + return b.backend.Rebind(query) +} + +func (b *SQLQueueBackend) boolLiteral(value bool) string { + if b.backend == nil { + if value { + return "1" + } + return "0" + } + return b.backend.BoolLiteral(value) +} + // getTableName retrieves the table name for a queue, using cache when possible // If forceRefresh is true, it will bypass the cache and query from database -func (b *TiDBBackend) getTableName(queueName string, forceRefresh bool) (string, error) { +func (b *SQLQueueBackend) getTableName(queueName string, forceRefresh bool) (string, error) { // Try to get from cache first (unless force refresh) if !forceRefresh { b.cacheMu.RLock() @@ -295,7 +338,7 @@ func (b *TiDBBackend) getTableName(queueName string, forceRefresh bool) (string, // Query from database var tableName string err := b.db.QueryRow( - "SELECT table_name FROM queuefs_registry WHERE queue_name = ?", + b.rebind("SELECT table_name FROM queuefs_registry WHERE queue_name = ?"), queueName, ).Scan(&tableName) @@ -312,13 +355,13 @@ func (b *TiDBBackend) getTableName(queueName string, forceRefresh bool) (string, } // invalidateCache removes a queue from the cache -func (b *TiDBBackend) invalidateCache(queueName string) { +func (b *SQLQueueBackend) invalidateCache(queueName string) { b.cacheMu.Lock() delete(b.tableCache, queueName) b.cacheMu.Unlock() } -func (b *TiDBBackend) Enqueue(queueName string, msg QueueMessage) error { +func (b *SQLQueueBackend) Enqueue(queueName string, msg QueueMessage) error { msgData, err := json.Marshal(msg) if err != nil { return fmt.Errorf("failed to marshal message: %w", err) @@ -334,9 +377,10 @@ func (b *TiDBBackend) Enqueue(queueName string, msg QueueMessage) error { // Insert message into queue table insertSQL := fmt.Sprintf( - "INSERT INTO %s (message_id, data, timestamp, deleted) VALUES (?, ?, ?, 0)", - tableName, + "INSERT INTO %s (message_id, data, timestamp, deleted) VALUES (?, ?, ?, %s)", + tableName, b.boolLiteral(false), ) + insertSQL = b.rebind(insertSQL) _, err = b.db.Exec(insertSQL, msg.ID, string(msgData), msg.Timestamp.Unix()) if err != nil { return fmt.Errorf("failed to enqueue message: %w", err) @@ -345,7 +389,7 @@ func (b *TiDBBackend) Enqueue(queueName string, msg QueueMessage) error { return nil } -func (b *TiDBBackend) Dequeue(queueName string) (QueueMessage, bool, error) { +func (b *SQLQueueBackend) Dequeue(queueName string) (QueueMessage, bool, error) { // Get table name from cache (lazy loading) tableName, err := b.getTableName(queueName, false) if err == sql.ErrNoRows { @@ -367,9 +411,13 @@ func (b *TiDBBackend) Dequeue(queueName string) (QueueMessage, bool, error) { var data string querySQL := fmt.Sprintf( - "SELECT id, data FROM %s WHERE deleted = 0 ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED", - tableName, + "SELECT id, data FROM %s WHERE deleted = %s ORDER BY id LIMIT 1", + tableName, b.boolLiteral(false), ) + if b.backend.SupportsSkipLocked() { + querySQL += " FOR UPDATE SKIP LOCKED" + } + querySQL = b.rebind(querySQL) err = tx.QueryRow(querySQL).Scan(&id, &data) if err == sql.ErrNoRows { @@ -380,13 +428,21 @@ func (b *TiDBBackend) Dequeue(queueName string) (QueueMessage, bool, error) { // Mark the message as deleted updateSQL := fmt.Sprintf( - "UPDATE %s SET deleted = 1, deleted_at = CURRENT_TIMESTAMP WHERE id = ?", - tableName, + "UPDATE %s SET deleted = %s, deleted_at = CURRENT_TIMESTAMP WHERE id = ? AND deleted = %s", + tableName, b.boolLiteral(true), b.boolLiteral(false), ) - _, err = tx.Exec(updateSQL, id) + updateSQL = b.rebind(updateSQL) + result, err := tx.Exec(updateSQL, id) if err != nil { return QueueMessage{}, false, fmt.Errorf("failed to mark message as deleted: %w", err) } + rowsAffected, err := result.RowsAffected() + if err != nil { + return QueueMessage{}, false, fmt.Errorf("failed to check dequeue result: %w", err) + } + if rowsAffected == 0 { + return QueueMessage{}, false, nil + } // Commit transaction if err := tx.Commit(); err != nil { @@ -402,7 +458,7 @@ func (b *TiDBBackend) Dequeue(queueName string) (QueueMessage, bool, error) { return msg, true, nil } -func (b *TiDBBackend) Peek(queueName string) (QueueMessage, bool, error) { +func (b *SQLQueueBackend) Peek(queueName string) (QueueMessage, bool, error) { // Get table name from cache (lazy loading) tableName, err := b.getTableName(queueName, false) if err == sql.ErrNoRows { @@ -413,8 +469,8 @@ func (b *TiDBBackend) Peek(queueName string) (QueueMessage, bool, error) { var data string querySQL := fmt.Sprintf( - "SELECT data FROM %s WHERE deleted = 0 ORDER BY id LIMIT 1", - tableName, + "SELECT data FROM %s WHERE deleted = %s ORDER BY id LIMIT 1", + tableName, b.boolLiteral(false), ) err = b.db.QueryRow(querySQL).Scan(&data) @@ -433,7 +489,7 @@ func (b *TiDBBackend) Peek(queueName string) (QueueMessage, bool, error) { return msg, true, nil } -func (b *TiDBBackend) Size(queueName string) (int, error) { +func (b *SQLQueueBackend) Size(queueName string) (int, error) { // Get table name from cache (lazy loading) tableName, err := b.getTableName(queueName, false) if err == sql.ErrNoRows { @@ -444,8 +500,8 @@ func (b *TiDBBackend) Size(queueName string) (int, error) { var count int querySQL := fmt.Sprintf( - "SELECT COUNT(*) FROM %s WHERE deleted = 0", - tableName, + "SELECT COUNT(*) FROM %s WHERE deleted = %s", + tableName, b.boolLiteral(false), ) err = b.db.QueryRow(querySQL).Scan(&count) if err != nil { @@ -454,7 +510,7 @@ func (b *TiDBBackend) Size(queueName string) (int, error) { return count, nil } -func (b *TiDBBackend) Clear(queueName string) error { +func (b *SQLQueueBackend) Clear(queueName string) error { // Get table name from cache (lazy loading) tableName, err := b.getTableName(queueName, false) if err == sql.ErrNoRows { @@ -472,7 +528,7 @@ func (b *TiDBBackend) Clear(queueName string) error { return nil } -func (b *TiDBBackend) ListQueues(prefix string) ([]string, error) { +func (b *SQLQueueBackend) ListQueues(prefix string) ([]string, error) { // Query from registry table to include all queues var query string var args []interface{} @@ -483,6 +539,7 @@ func (b *TiDBBackend) ListQueues(prefix string) ([]string, error) { query = "SELECT queue_name FROM queuefs_registry WHERE queue_name = ? OR queue_name LIKE ?" args = []interface{}{prefix, prefix + "/%"} } + query = b.rebind(query) rows, err := b.db.Query(query, args...) if err != nil { @@ -502,7 +559,7 @@ func (b *TiDBBackend) ListQueues(prefix string) ([]string, error) { return queues, nil } -func (b *TiDBBackend) GetLastEnqueueTime(queueName string) (time.Time, error) { +func (b *SQLQueueBackend) GetLastEnqueueTime(queueName string) (time.Time, error) { // Get table name from cache (lazy loading) tableName, err := b.getTableName(queueName, false) if err == sql.ErrNoRows { @@ -511,23 +568,27 @@ func (b *TiDBBackend) GetLastEnqueueTime(queueName string) (time.Time, error) { return time.Time{}, fmt.Errorf("failed to get queue table name: %w", err) } - var timestamp int64 + var timestamp sql.NullInt64 querySQL := fmt.Sprintf( - "SELECT MAX(timestamp) FROM %s WHERE deleted = 0", - tableName, + "SELECT MAX(timestamp) FROM %s WHERE deleted = %s", + tableName, b.boolLiteral(false), ) err = b.db.QueryRow(querySQL).Scan(×tamp) - if err == sql.ErrNoRows || timestamp == 0 { - return time.Time{}, nil - } else if err != nil { + if err != nil { + if err == sql.ErrNoRows { + return time.Time{}, nil + } return time.Time{}, fmt.Errorf("failed to get last enqueue time: %w", err) } + if !timestamp.Valid || timestamp.Int64 == 0 { + return time.Time{}, nil + } - return time.Unix(timestamp, 0), nil + return time.Unix(timestamp.Int64, 0), nil } -func (b *TiDBBackend) RemoveQueue(queueName string) error { +func (b *SQLQueueBackend) RemoveQueue(queueName string) error { if queueName == "" { // Remove all queues: drop all queue tables and clear registry rows, err := b.db.Query("SELECT queue_name, table_name FROM queuefs_registry") @@ -572,7 +633,7 @@ func (b *TiDBBackend) RemoveQueue(queueName string) error { // Remove queue and nested queues rows, err := b.db.Query( - "SELECT queue_name, table_name FROM queuefs_registry WHERE queue_name = ? OR queue_name LIKE ?", + b.rebind("SELECT queue_name, table_name FROM queuefs_registry WHERE queue_name = ? OR queue_name LIKE ?"), queueName, queueName+"/%", ) if err != nil { @@ -610,27 +671,29 @@ func (b *TiDBBackend) RemoveQueue(queueName string) error { // Remove from registry _, err = b.db.Exec( - "DELETE FROM queuefs_registry WHERE queue_name = ? OR queue_name LIKE ?", + b.rebind("DELETE FROM queuefs_registry WHERE queue_name = ? OR queue_name LIKE ?"), queueName, queueName+"/%", ) return err } -func (b *TiDBBackend) CreateQueue(queueName string) error { +func (b *SQLQueueBackend) CreateQueue(queueName string) error { // Generate table name tableName := sanitizeTableName(queueName) // Create the queue table - createTableSQL := getCreateTableSQL(tableName) + createTableSQL := b.backend.QueueTableDDL(tableName) if _, err := b.db.Exec(createTableSQL); err != nil { return fmt.Errorf("failed to create queue table: %w", err) } + if err := b.backend.EnsureQueueIndexes(b.db, tableName); err != nil { + return err + } // Register in queuefs_registry - _, err := b.db.Exec( - "INSERT IGNORE INTO queuefs_registry (queue_name, table_name) VALUES (?, ?)", - queueName, tableName, - ) + registerSQL := b.backend.RegistryInsertSQL() + registerSQL = b.rebind(registerSQL) + _, err := b.db.Exec(registerSQL, queueName, tableName) if err != nil { return fmt.Errorf("failed to register queue: %w", err) } @@ -644,7 +707,7 @@ func (b *TiDBBackend) CreateQueue(queueName string) error { return nil } -func (b *TiDBBackend) QueueExists(queueName string) (bool, error) { +func (b *SQLQueueBackend) QueueExists(queueName string) (bool, error) { // Check cache first b.cacheMu.RLock() _, exists := b.tableCache[queueName] @@ -657,7 +720,7 @@ func (b *TiDBBackend) QueueExists(queueName string) (bool, error) { // If not in cache, query database var count int err := b.db.QueryRow( - "SELECT COUNT(*) FROM queuefs_registry WHERE queue_name = ?", + b.rebind("SELECT COUNT(*) FROM queuefs_registry WHERE queue_name = ?"), queueName, ).Scan(&count) if err != nil { diff --git a/agfs-server/pkg/plugins/queuefs/db_backend.go b/agfs-server/pkg/plugins/queuefs/db_backend.go index 03b7342..6574cce 100644 --- a/agfs-server/pkg/plugins/queuefs/db_backend.go +++ b/agfs-server/pkg/plugins/queuefs/db_backend.go @@ -4,12 +4,15 @@ import ( "crypto/tls" "database/sql" "fmt" + "net/url" "regexp" + "strconv" "strings" "github.com/c4pt0r/agfs/agfs-server/pkg/plugin/config" "github.com/go-sql-driver/mysql" _ "github.com/go-sql-driver/mysql" // MySQL/TiDB driver + _ "github.com/jackc/pgx/v5/stdlib" // PostgreSQL driver _ "github.com/mattn/go-sqlite3" // SQLite driver log "github.com/sirupsen/logrus" ) @@ -22,8 +25,23 @@ type DBBackend interface { // GetInitSQL returns the SQL statements to initialize the schema GetInitSQL() []string - // GetDriverName returns the driver name - GetDriverName() string + // SupportsSkipLocked reports whether the backend supports FOR UPDATE SKIP LOCKED. + SupportsSkipLocked() bool + + // QueueTableDDL returns the SQL to create a queue table. + QueueTableDDL(tableName string) string + + // EnsureQueueIndexes creates any backend-specific queue indexes. + EnsureQueueIndexes(db *sql.DB, tableName string) error + + // RegistryInsertSQL returns the SQL used to register a queue. + RegistryInsertSQL() string + + // Rebind rewrites generic placeholders to the backend-specific format. + Rebind(query string) string + + // BoolLiteral returns the backend-specific SQL literal for a boolean value. + BoolLiteral(value bool) string } // SQLiteDBBackend implements DBBackend for SQLite @@ -33,10 +51,6 @@ func NewSQLiteDBBackend() *SQLiteDBBackend { return &SQLiteDBBackend{} } -func (b *SQLiteDBBackend) GetDriverName() string { - return "sqlite3" -} - func (b *SQLiteDBBackend) Open(cfg map[string]interface{}) (*sql.DB, error) { dbPath := config.GetStringConfig(cfg, "db_path", "queue.db") @@ -56,26 +70,58 @@ func (b *SQLiteDBBackend) Open(cfg map[string]interface{}) (*sql.DB, error) { func (b *SQLiteDBBackend) GetInitSQL() []string { return []string{ - // Queue metadata table to track all queues (including empty ones) - `CREATE TABLE IF NOT EXISTS queue_metadata ( + `CREATE TABLE IF NOT EXISTS queuefs_registry ( queue_name TEXT PRIMARY KEY, - created_at INTEGER DEFAULT (strftime('%s', 'now')), - last_updated INTEGER DEFAULT (strftime('%s', 'now')) - )`, - // Queue messages table - `CREATE TABLE IF NOT EXISTS queue_messages ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - queue_name TEXT NOT NULL, - message_id TEXT NOT NULL, - data TEXT NOT NULL, - timestamp INTEGER NOT NULL, - created_at INTEGER DEFAULT (strftime('%s', 'now')) + table_name TEXT NOT NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP )`, - `CREATE INDEX IF NOT EXISTS idx_queue_name ON queue_messages(queue_name)`, - `CREATE INDEX IF NOT EXISTS idx_queue_order ON queue_messages(queue_name, id)`, } } +func (b *SQLiteDBBackend) SupportsSkipLocked() bool { + return false +} + +func (b *SQLiteDBBackend) QueueTableDDL(tableName string) string { + return fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + message_id TEXT NOT NULL, + data BLOB NOT NULL, + timestamp INTEGER NOT NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + deleted INTEGER DEFAULT 0, + deleted_at DATETIME NULL + )`, tableName) +} + +func (b *SQLiteDBBackend) EnsureQueueIndexes(db *sql.DB, tableName string) error { + indexSQL := fmt.Sprintf( + "CREATE INDEX IF NOT EXISTS idx_%s_deleted_id ON %s(deleted, id)", + strings.TrimPrefix(tableName, "queuefs_queue_"), + tableName, + ) + _, err := db.Exec(indexSQL) + if err != nil { + return fmt.Errorf("failed to create queue index: %w", err) + } + return nil +} + +func (b *SQLiteDBBackend) RegistryInsertSQL() string { + return "INSERT OR IGNORE INTO queuefs_registry (queue_name, table_name) VALUES (?, ?)" +} + +func (b *SQLiteDBBackend) Rebind(query string) string { + return query +} + +func (b *SQLiteDBBackend) BoolLiteral(value bool) string { + if value { + return "1" + } + return "0" +} + // TiDBDBBackend implements DBBackend for TiDB type TiDBDBBackend struct{} @@ -83,10 +129,6 @@ func NewTiDBDBBackend() *TiDBDBBackend { return &TiDBDBBackend{} } -func (b *TiDBDBBackend) GetDriverName() string { - return "mysql" -} - func (b *TiDBDBBackend) Open(cfg map[string]interface{}) (*sql.DB, error) { // Check if DSN contains tls parameter dsnStr := config.GetStringConfig(cfg, "dsn", "") @@ -210,6 +252,214 @@ func (b *TiDBDBBackend) GetInitSQL() []string { } } +func (b *TiDBDBBackend) SupportsSkipLocked() bool { + return true +} + +func (b *TiDBDBBackend) QueueTableDDL(tableName string) string { + return fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + message_id VARCHAR(64) NOT NULL, + data LONGBLOB NOT NULL, + timestamp BIGINT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + deleted TINYINT(1) DEFAULT 0, + deleted_at TIMESTAMP NULL, + INDEX idx_deleted_id (deleted, id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`, tableName) +} + +func (b *TiDBDBBackend) EnsureQueueIndexes(db *sql.DB, tableName string) error { + return nil +} + +func (b *TiDBDBBackend) RegistryInsertSQL() string { + return "INSERT IGNORE INTO queuefs_registry (queue_name, table_name) VALUES (?, ?)" +} + +func (b *TiDBDBBackend) Rebind(query string) string { + return query +} + +func (b *TiDBDBBackend) BoolLiteral(value bool) string { + if value { + return "1" + } + return "0" +} + +// PostgreSQLDBBackend implements DBBackend for PostgreSQL. +type PostgreSQLDBBackend struct{} + +func NewPostgreSQLDBBackend() *PostgreSQLDBBackend { + return &PostgreSQLDBBackend{} +} + +func (b *PostgreSQLDBBackend) Open(cfg map[string]interface{}) (*sql.DB, error) { + dsn := config.GetStringConfig(cfg, "dsn", "") + database := config.GetStringConfig(cfg, "database", "") + if dsn == "" { + if database == "" { + database = "queuedb" + } + host := config.GetStringConfig(cfg, "host", "127.0.0.1") + port := config.GetStringConfig(cfg, "port", "5432") + user := config.GetStringConfig(cfg, "user", "postgres") + password := config.GetStringConfig(cfg, "password", "") + sslMode := "disable" + if config.GetBoolConfig(cfg, "enable_tls", false) { + sslMode = "require" + if config.GetBoolConfig(cfg, "tls_skip_verify", false) { + sslMode = "prefer" + } + } + + dsn = fmt.Sprintf("host=%s port=%s user=%s dbname=%s sslmode=%s", + host, port, user, database, sslMode) + if password != "" { + dsn += fmt.Sprintf(" password=%s", password) + } + if serverName := config.GetStringConfig(cfg, "tls_server_name", ""); serverName != "" { + dsn += fmt.Sprintf(" host=%s", serverName) + } + } + + parsedDSN, err := url.Parse(dsn) + if err == nil { + if database == "" { + database = strings.TrimPrefix(parsedDSN.Path, "/") + } + if database == "" { + database = "postgres" + } + parsedDSN.Path = "/" + database + dsn = parsedDSN.String() + } else if database == "" { + database = "queuedb" + } + + adminDSN := config.GetStringConfig(cfg, "admin_dsn", "") + if adminDSN == "" && database != "" { + if parsedDSN != nil { + adminURL := *parsedDSN + adminURL.Path = "/postgres" + adminDSN = adminURL.String() + } else { + host := config.GetStringConfig(cfg, "host", "127.0.0.1") + port := config.GetStringConfig(cfg, "port", "5432") + user := config.GetStringConfig(cfg, "user", "postgres") + password := config.GetStringConfig(cfg, "password", "") + sslMode := "disable" + if config.GetBoolConfig(cfg, "enable_tls", false) { + sslMode = "require" + if config.GetBoolConfig(cfg, "tls_skip_verify", false) { + sslMode = "prefer" + } + } + adminDSN = fmt.Sprintf("host=%s port=%s user=%s dbname=postgres sslmode=%s", + host, port, user, sslMode) + if password != "" { + adminDSN += fmt.Sprintf(" password=%s", password) + } + } + } + + if adminDSN != "" && database != "" { + tempDB, err := sql.Open("pgx", adminDSN) + if err == nil { + defer tempDB.Close() + _, err = tempDB.Exec(fmt.Sprintf("CREATE DATABASE %s", quotePostgresIdentifier(database))) + if err != nil && !strings.Contains(err.Error(), "already exists") { + log.Warnf("[queuefs] Failed to create PostgreSQL database %q: %v", database, err) + } else { + log.Infof("[queuefs] PostgreSQL database %q created or already exists", database) + } + } + } + + db, err := sql.Open("pgx", dsn) + if err != nil { + return nil, fmt.Errorf("failed to open PostgreSQL database: %w", err) + } + + db.SetMaxOpenConns(100) + db.SetMaxIdleConns(10) + + if err := db.Ping(); err != nil { + db.Close() + return nil, fmt.Errorf("failed to ping PostgreSQL database: %w", err) + } + + return db, nil +} + +func (b *PostgreSQLDBBackend) GetInitSQL() []string { + return []string{ + `CREATE TABLE IF NOT EXISTS queuefs_registry ( + queue_name TEXT PRIMARY KEY, + table_name TEXT NOT NULL, + created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP + )`, + } +} + +func (b *PostgreSQLDBBackend) SupportsSkipLocked() bool { + return true +} + +func (b *PostgreSQLDBBackend) QueueTableDDL(tableName string) string { + return fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s ( + id BIGSERIAL PRIMARY KEY, + message_id TEXT NOT NULL, + data BYTEA NOT NULL, + timestamp BIGINT NOT NULL, + created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP, + deleted BOOLEAN DEFAULT FALSE, + deleted_at TIMESTAMPTZ NULL + )`, tableName) +} + +func (b *PostgreSQLDBBackend) EnsureQueueIndexes(db *sql.DB, tableName string) error { + indexName := fmt.Sprintf("idx_%s_deleted_id", strings.TrimPrefix(tableName, "queuefs_queue_")) + indexSQL := fmt.Sprintf( + "CREATE INDEX IF NOT EXISTS %s ON %s(deleted, id)", + quotePostgresIdentifier(indexName), + tableName, + ) + _, err := db.Exec(indexSQL) + if err != nil { + return fmt.Errorf("failed to create queue index: %w", err) + } + return nil +} + +func (b *PostgreSQLDBBackend) RegistryInsertSQL() string { + return `INSERT INTO queuefs_registry (queue_name, table_name) VALUES (?, ?) ON CONFLICT (queue_name) DO NOTHING` +} + +func (b *PostgreSQLDBBackend) Rebind(query string) string { + var builder strings.Builder + builder.Grow(len(query) + 8) + argIndex := 1 + for _, ch := range query { + if ch == '?' { + builder.WriteByte('$') + builder.WriteString(strconv.Itoa(argIndex)) + argIndex++ + continue + } + builder.WriteRune(ch) + } + return builder.String() +} + +func (b *PostgreSQLDBBackend) BoolLiteral(value bool) string { + if value { + return "TRUE" + } + return "FALSE" +} + // Helper functions func extractDatabaseName(dsn string, configDB string) string { @@ -241,18 +491,8 @@ func sanitizeTableName(queueName string) string { return "queuefs_queue_" + tableName } -// getCreateTableSQL returns the SQL to create a queue table -func getCreateTableSQL(tableName string) string { - return fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s ( - id BIGINT AUTO_INCREMENT PRIMARY KEY, - message_id VARCHAR(64) NOT NULL, - data LONGBLOB NOT NULL, - timestamp BIGINT NOT NULL, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - deleted TINYINT(1) DEFAULT 0, - deleted_at TIMESTAMP NULL, - INDEX idx_deleted_id (deleted, id) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`, tableName) +func quotePostgresIdentifier(identifier string) string { + return `"` + strings.ReplaceAll(identifier, `"`, `""`) + `"` } // CreateBackend creates the appropriate database backend @@ -264,6 +504,8 @@ func CreateBackend(cfg map[string]interface{}) (DBBackend, error) { return NewSQLiteDBBackend(), nil case "tidb", "mysql": return NewTiDBDBBackend(), nil + case "pgsql", "postgres", "postgresql": + return NewPostgreSQLDBBackend(), nil default: return nil, fmt.Errorf("unsupported database backend: %s", backendType) } diff --git a/agfs-server/pkg/plugins/queuefs/pgsql_backend_test.go b/agfs-server/pkg/plugins/queuefs/pgsql_backend_test.go new file mode 100644 index 0000000..4e26293 --- /dev/null +++ b/agfs-server/pkg/plugins/queuefs/pgsql_backend_test.go @@ -0,0 +1,269 @@ +package queuefs + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/url" + "os" + "sync" + "testing" + "time" +) + +func pgTestConfig(t *testing.T, database string) map[string]interface{} { + t.Helper() + + if os.Getenv("PG_TEST") == "" { + t.Skip("set PG_TEST=1 to run PostgreSQL integration tests") + } + + dsn := os.Getenv("PG_TEST_DSN") + if dsn == "" { + t.Skip("set PG_TEST_DSN to run PostgreSQL integration tests") + } + + parsedURL, err := url.Parse(dsn) + if err != nil { + t.Fatalf("parse PG_TEST_DSN: %v", err) + } + parsedURL.Path = "/" + database + query := parsedURL.Query() + if query.Get("sslmode") == "" { + query.Set("sslmode", "require") + } + parsedURL.RawQuery = query.Encode() + + adminURL := *parsedURL + adminURL.Path = "/postgres" + + adminDSN := adminURL.String() + password, _ := parsedURL.User.Password() + user := parsedURL.User.Username() + port := parsedURL.Port() + host := parsedURL.Hostname() + if port == "" { + port = "5432" + } + + return map[string]interface{}{ + "backend": "pgsql", + "dsn": parsedURL.String(), + "host": host, + "port": port, + "user": user, + "password": password, + "database": database, + "admin_dsn": adminDSN, + } +} + +func newPGTestQueueFS(t *testing.T, database string) *queueFS { + t.Helper() + + plugin := NewQueueFSPlugin() + if err := plugin.Initialize(pgTestConfig(t, database)); err != nil { + t.Fatalf("initialize pgsql queuefs: %v", err) + } + t.Cleanup(func() { + if plugin.backend != nil { + _ = plugin.backend.Close() + } + }) + + fs, ok := plugin.GetFileSystem().(*queueFS) + if !ok { + t.Fatalf("unexpected filesystem type %T", plugin.GetFileSystem()) + } + return fs +} + +func newPGTestDatabaseName() string { + return fmt.Sprintf("queuefs_pg_test_%d", time.Now().UnixNano()) +} + +func TestQueueFSPGSQLFileRegression(t *testing.T) { + database := newPGTestDatabaseName() + fs := newPGTestQueueFS(t, database) + + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if err := fs.Mkdir("/logs/errors", 0o755); err != nil { + t.Fatalf("mkdir /logs/errors: %v", err) + } + + entries, err := fs.ReadDir("/") + if err != nil { + t.Fatalf("readdir root: %v", err) + } + rootEntries := queueDirEntryNames(entries) + for _, name := range []string{"README", "jobs", "logs"} { + if _, ok := rootEntries[name]; !ok { + t.Fatalf("root missing %q in %+v", name, entries) + } + } + + if _, err := fs.Write("/jobs/enqueue", []byte("first"), -1, 0); err != nil { + t.Fatalf("enqueue first: %v", err) + } + if _, err := fs.Write("/jobs/enqueue", []byte("second"), -1, 0); err != nil { + t.Fatalf("enqueue second: %v", err) + } + + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "2" { + t.Fatalf("queue size = %q, want 2", got) + } + + peeked := mustReadMessage(t, fs, "/jobs/peek") + if peeked.Data != "first" { + t.Fatalf("peeked message = %q, want first", peeked.Data) + } + + first := mustReadMessage(t, fs, "/jobs/dequeue") + second := mustReadMessage(t, fs, "/jobs/dequeue") + if first.Data != "first" || second.Data != "second" { + t.Fatalf("dequeue order = [%q, %q], want [first, second]", first.Data, second.Data) + } + + if got := string(mustReadAll(t, fs, "/jobs/dequeue")); got != "{}" { + t.Fatalf("empty dequeue = %q, want {}", got) + } + + if _, err := fs.Write("/jobs/enqueue", []byte("to-clear"), -1, 0); err != nil { + t.Fatalf("enqueue before clear: %v", err) + } + if _, err := fs.Write("/jobs/clear", nil, -1, 0); err != nil { + t.Fatalf("clear queue: %v", err) + } + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "0" { + t.Fatalf("queue size after clear = %q, want 0", got) + } + + if err := fs.RemoveAll("/logs"); err != nil { + t.Fatalf("removeall /logs: %v", err) + } + if _, err := fs.Stat("/logs/errors"); err == nil { + t.Fatal("expected removed nested pgsql queue to disappear") + } +} + +func TestQueueFSPGSQLPersistenceRegression(t *testing.T) { + database := newPGTestDatabaseName() + + func() { + fs := newPGTestQueueFS(t, database) + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if _, err := fs.Write("/jobs/enqueue", []byte("persisted"), -1, 0); err != nil { + t.Fatalf("enqueue persisted message: %v", err) + } + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "1" { + t.Fatalf("initial queue size = %q, want 1", got) + } + }() + + fs := newPGTestQueueFS(t, database) + + entries, err := fs.ReadDir("/") + if err != nil { + t.Fatalf("readdir root after reopen: %v", err) + } + if _, ok := queueDirEntryNames(entries)["jobs"]; !ok { + t.Fatalf("root missing reopened queue in %+v", entries) + } + + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "1" { + t.Fatalf("reopened queue size = %q, want 1", got) + } + peeked := mustReadMessage(t, fs, "/jobs/peek") + if peeked.Data != "persisted" { + t.Fatalf("peek after reopen = %q, want persisted", peeked.Data) + } + + dequeued := mustReadMessage(t, fs, "/jobs/dequeue") + if dequeued.Data != "persisted" { + t.Fatalf("dequeue after reopen = %q, want persisted", dequeued.Data) + } + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "0" { + t.Fatalf("queue size after reopened dequeue = %q, want 0", got) + } + + if _, err := fs.Stat("/jobs"); err != nil { + t.Fatalf("stat empty queue after reopen: %v", err) + } +} + +func TestQueueFSPGSQLConfigUsesDSN(t *testing.T) { + if os.Getenv("PG_TEST") == "" { + t.Skip("set PG_TEST=1 to run PostgreSQL integration tests") + } + + config := pgTestConfig(t, newPGTestDatabaseName()) + if _, ok := config["dsn"].(string); !ok || config["dsn"] == "" { + t.Fatalf("expected non-empty dsn in config: %+v", config) + } +} + +func TestQueueFSPGSQLConcurrentDequeueRegression(t *testing.T) { + database := newPGTestDatabaseName() + writerFS := newPGTestQueueFS(t, database) + readerOne := newPGTestQueueFS(t, database) + readerTwo := newPGTestQueueFS(t, database) + + if err := writerFS.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if _, err := writerFS.Write("/jobs/enqueue", []byte("once"), -1, 0); err != nil { + t.Fatalf("enqueue once: %v", err) + } + + type dequeueResult struct { + payload []byte + err error + } + + start := make(chan struct{}) + results := make(chan dequeueResult, 2) + var wg sync.WaitGroup + for _, fs := range []*queueFS{readerOne, readerTwo} { + wg.Add(1) + go func(fs *queueFS) { + defer wg.Done() + <-start + payload, err := fs.Read("/jobs/dequeue", 0, -1) + results <- dequeueResult{payload: payload, err: err} + }(fs) + } + close(start) + wg.Wait() + close(results) + + nonEmpty := 0 + for result := range results { + if result.err != nil && !errors.Is(result.err, io.EOF) { + t.Fatalf("concurrent dequeue: %v", result.err) + } + if string(result.payload) == "{}" { + continue + } + + var msg QueueMessage + if err := json.Unmarshal(result.payload, &msg); err != nil { + t.Fatalf("unmarshal concurrent dequeue payload: %v (payload=%q)", err, string(result.payload)) + } + if msg.Data != "once" { + t.Fatalf("concurrent dequeue returned %q, want once", msg.Data) + } + nonEmpty++ + } + + if nonEmpty != 1 { + t.Fatalf("concurrent dequeue claimed %d messages, want 1", nonEmpty) + } + if got := string(mustReadAll(t, writerFS, "/jobs/size")); got != "0" { + t.Fatalf("queue size after concurrent dequeue = %q, want 0", got) + } +} diff --git a/agfs-server/pkg/plugins/queuefs/queuefs.go b/agfs-server/pkg/plugins/queuefs/queuefs.go index 521bb5b..f50d2c4 100644 --- a/agfs-server/pkg/plugins/queuefs/queuefs.go +++ b/agfs-server/pkg/plugins/queuefs/queuefs.go @@ -43,6 +43,7 @@ const ( // - memory (default): In-memory storage // - tidb: TiDB database storage with TLS support // - sqlite: SQLite database storage +// - pgsql: PostgreSQL database storage type QueueFSPlugin struct { backend QueueBackend mu sync.RWMutex // Protects backend operations @@ -83,7 +84,7 @@ func (q *QueueFSPlugin) Validate(cfg map[string]interface{}) error { allowedKeys := []string{ "backend", "mount_path", // Database-related keys - "db_path", "dsn", "user", "password", "host", "port", "database", + "db_path", "dsn", "admin_dsn", "user", "password", "host", "port", "database", "enable_tls", "tls_server_name", "tls_skip_verify", } if err := config.ValidateOnlyKnownKeys(cfg, allowedKeys); err != nil { @@ -93,19 +94,22 @@ func (q *QueueFSPlugin) Validate(cfg map[string]interface{}) error { // Validate backend type backendType := config.GetStringConfig(cfg, "backend", "memory") validBackends := map[string]bool{ - "memory": true, - "tidb": true, - "mysql": true, - "sqlite": true, - "sqlite3": true, + "memory": true, + "tidb": true, + "mysql": true, + "pgsql": true, + "postgres": true, + "postgresql": true, + "sqlite": true, + "sqlite3": true, } if !validBackends[backendType] { - return fmt.Errorf("unsupported backend: %s (valid options: memory, tidb, mysql, sqlite)", backendType) + return fmt.Errorf("unsupported backend: %s (valid options: memory, tidb, mysql, pgsql, sqlite)", backendType) } // Validate database-related parameters if backend is not memory if backendType != "memory" { - for _, key := range []string{"db_path", "dsn", "user", "password", "host", "database", "tls_server_name"} { + for _, key := range []string{"db_path", "dsn", "admin_dsn", "user", "password", "host", "database", "tls_server_name"} { if err := config.ValidateStringType(cfg, key); err != nil { return err } @@ -137,8 +141,12 @@ func (q *QueueFSPlugin) Initialize(cfg map[string]interface{}) error { switch backendType { case "memory": backend = NewMemoryBackend() - case "tidb", "mysql", "sqlite", "sqlite3": - backend = NewTiDBBackend() + case "sqlite", "sqlite3": + backend = NewSQLiteQueueBackend() + case "pgsql", "postgres", "postgresql": + backend = NewPostgresQueueBackend() + case "tidb", "mysql": + backend = NewTiDBQueueBackend() default: return fmt.Errorf("unsupported backend: %s", backendType) } @@ -205,44 +213,57 @@ NESTED QUEUES: BACKENDS: Memory Backend (default): - [plugins.queuefs] - enabled = true - path = "/queuefs" - # No additional config needed for memory backend + [plugins.queuefs] + enabled = true + path = "/queuefs" + # No additional config needed for memory backend SQLite Backend: - [plugins.queuefs] - enabled = true - path = "/queuefs" + [plugins.queuefs] + enabled = true + path = "/queuefs" [plugins.queuefs.config] backend = "sqlite" db_path = "queue.db" + PostgreSQL Backend (local): + [plugins.queuefs] + enabled = true + path = "/queuefs" + + [plugins.queuefs.config] + backend = "pgsql" + host = "127.0.0.1" + port = 5432 + user = "postgres" + password = "" + database = "queuedb" + TiDB Backend (local): - [plugins.queuefs] - enabled = true - path = "/queuefs" + [plugins.queuefs] + enabled = true + path = "/queuefs" [plugins.queuefs.config] backend = "tidb" host = "127.0.0.1" - port = "4000" + port = 4000 user = "root" password = "" database = "queuedb" TiDB Cloud Backend (with TLS): - [plugins.queuefs] - enabled = true - path = "/queuefs" + [plugins.queuefs] + enabled = true + path = "/queuefs" [plugins.queuefs.config] backend = "tidb" user = "3YdGXuXNdAEmP1f.root" password = "your_password" host = "gateway01.us-west-2.prod.aws.tidbcloud.com" - port = "4000" + port = 4000 database = "queuedb" enable_tls = true tls_server_name = "gateway01.us-west-2.prod.aws.tidbcloud.com" @@ -276,6 +297,7 @@ EXAMPLES: BACKEND COMPARISON: - memory: Fastest, no persistence, lost on restart - sqlite: Good for single server, persistent, file-based + - pgsql: Good for local and server deployments, persistent, transactional - tidb: Best for production, distributed, scalable, persistent ` } @@ -287,7 +309,7 @@ func (q *QueueFSPlugin) GetConfigParams() []plugin.ConfigParameter { Type: "string", Required: false, Default: "memory", - Description: "Queue backend (memory, tidb, mysql, sqlite, sqlite3)", + Description: "Queue backend (memory, tidb, mysql, pgsql, postgres, sqlite, sqlite3)", }, { Name: "db_path", @@ -303,6 +325,13 @@ func (q *QueueFSPlugin) GetConfigParams() []plugin.ConfigParameter { Default: "", Description: "Database connection string (DSN)", }, + { + Name: "admin_dsn", + Type: "string", + Required: false, + Default: "", + Description: "Administrative DSN used to create the target database when it does not exist", + }, { Name: "user", Type: "string", @@ -1028,9 +1057,9 @@ type queueFileHandle struct { // handleManager manages open handles for queueFS type handleManager struct { - handles map[int64]*queueFileHandle - nextID int64 - mu sync.Mutex + handles map[int64]*queueFileHandle + nextID int64 + mu sync.Mutex } // Global handle manager for queueFS (per plugin instance would be better, but keeping it simple) diff --git a/agfs-server/pkg/plugins/queuefs/queuefs_contract_test.go b/agfs-server/pkg/plugins/queuefs/queuefs_contract_test.go new file mode 100644 index 0000000..544722a --- /dev/null +++ b/agfs-server/pkg/plugins/queuefs/queuefs_contract_test.go @@ -0,0 +1,275 @@ +package queuefs + +import ( + "encoding/json" + "errors" + "io" + "strings" + "testing" + + "github.com/c4pt0r/agfs/agfs-server/pkg/filesystem" +) + +func mustReadAll(t *testing.T, fs filesystem.FileSystem, path string) []byte { + t.Helper() + + data, err := fs.Read(path, 0, -1) + if err != nil && !errors.Is(err, io.EOF) { + t.Fatalf("read %s: %v", path, err) + } + return data +} + +func newTestQueueFS(t *testing.T) *queueFS { + t.Helper() + + plugin := NewQueueFSPlugin() + if err := plugin.Initialize(map[string]interface{}{"backend": "memory"}); err != nil { + t.Fatalf("initialize queuefs: %v", err) + } + t.Cleanup(func() { + if plugin.backend != nil { + _ = plugin.backend.Close() + } + }) + + fs, ok := plugin.GetFileSystem().(*queueFS) + if !ok { + t.Fatalf("unexpected filesystem type %T", plugin.GetFileSystem()) + } + return fs +} + +func queueDirEntryNames(entries []filesystem.FileInfo) map[string]filesystem.FileInfo { + byName := make(map[string]filesystem.FileInfo, len(entries)) + for _, entry := range entries { + byName[entry.Name] = entry + } + return byName +} + +func mustReadMessage(t *testing.T, fs filesystem.FileSystem, path string) QueueMessage { + t.Helper() + + data, err := fs.Read(path, 0, -1) + if err != nil && !errors.Is(err, io.EOF) { + t.Fatalf("read %s: %v", path, err) + } + + var msg QueueMessage + if err := json.Unmarshal(data, &msg); err != nil { + t.Fatalf("unmarshal %s: %v (payload=%q)", path, err, string(data)) + } + return msg +} + +func TestQueueFSRootAndNestedQueueRegression(t *testing.T) { + fs := newTestQueueFS(t) + + rootInfo, err := fs.Stat("/") + if err != nil { + t.Fatalf("stat root: %v", err) + } + if !rootInfo.IsDir { + t.Fatalf("root should be a directory") + } + if got := rootInfo.Meta.Content["backend"]; got != "memory" { + t.Fatalf("root backend = %q, want memory", got) + } + + readme := mustReadAll(t, fs, "/README") + if !strings.Contains(string(readme), "QueueFS Plugin") { + t.Fatalf("README missing plugin description: %q", string(readme)) + } + + entries, err := fs.ReadDir("/") + if err != nil { + t.Fatalf("readdir root: %v", err) + } + rootEntries := queueDirEntryNames(entries) + if len(rootEntries) != 1 || rootEntries["README"].Name != "README" { + t.Fatalf("unexpected initial root entries: %+v", entries) + } + + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if err := fs.Mkdir("/logs/errors", 0o755); err != nil { + t.Fatalf("mkdir /logs/errors: %v", err) + } + + entries, err = fs.ReadDir("/") + if err != nil { + t.Fatalf("readdir root after mkdir: %v", err) + } + rootEntries = queueDirEntryNames(entries) + for _, name := range []string{"README", "jobs", "logs"} { + if _, ok := rootEntries[name]; !ok { + t.Fatalf("root missing %q in %+v", name, entries) + } + } + if !rootEntries["jobs"].IsDir || !rootEntries["logs"].IsDir { + t.Fatalf("expected queue directories at root: %+v", entries) + } + + jobsEntries, err := fs.ReadDir("/jobs") + if err != nil { + t.Fatalf("readdir /jobs: %v", err) + } + got := queueDirEntryNames(jobsEntries) + if len(got) != 5 { + t.Fatalf("unexpected /jobs control files: %+v", jobsEntries) + } + for _, name := range []string{"enqueue", "dequeue", "peek", "size", "clear"} { + if _, ok := got[name]; !ok { + t.Fatalf("/jobs missing control file %q in %+v", name, jobsEntries) + } + } + + logsEntries, err := fs.ReadDir("/logs") + if err != nil { + t.Fatalf("readdir /logs: %v", err) + } + logChildren := queueDirEntryNames(logsEntries) + if len(logChildren) != 1 || !logChildren["errors"].IsDir { + t.Fatalf("unexpected /logs entries: %+v", logsEntries) + } + + errorsInfo, err := fs.Stat("/logs/errors") + if err != nil { + t.Fatalf("stat /logs/errors: %v", err) + } + if !errorsInfo.IsDir { + t.Fatalf("/logs/errors should be a directory") + } + + if err := fs.RemoveAll("/logs"); err != nil { + t.Fatalf("removeall /logs: %v", err) + } + if _, err := fs.Stat("/logs"); err == nil || !strings.Contains(err.Error(), "no such file or directory") { + t.Fatalf("stat removed /logs error = %v, want missing path", err) + } + if _, err := fs.Stat("/logs/errors"); err == nil || !strings.Contains(err.Error(), "no such file or directory") { + t.Fatalf("stat removed /logs/errors error = %v, want missing path", err) + } +} + +func TestQueueFSControlFileRegression(t *testing.T) { + fs := newTestQueueFS(t) + + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + + if _, err := fs.Write("/jobs/enqueue", []byte("first"), -1, filesystem.WriteFlagCreate|filesystem.WriteFlagTruncate); err != nil { + t.Fatalf("enqueue first: %v", err) + } + if _, err := fs.Write("/jobs/enqueue", []byte("second"), -1, filesystem.WriteFlagAppend); err != nil { + t.Fatalf("enqueue second: %v", err) + } + + sizeData := mustReadAll(t, fs, "/jobs/size") + if got := string(sizeData); got != "2" { + t.Fatalf("queue size = %q, want 2", got) + } + + peeked := mustReadMessage(t, fs, "/jobs/peek") + if peeked.Data != "first" { + t.Fatalf("peeked message = %q, want first", peeked.Data) + } + + first := mustReadMessage(t, fs, "/jobs/dequeue") + if first.Data != "first" { + t.Fatalf("first dequeue = %q, want first", first.Data) + } + + sizeData = mustReadAll(t, fs, "/jobs/size") + if got := string(sizeData); got != "1" { + t.Fatalf("queue size after first dequeue = %q, want 1", got) + } + + second := mustReadMessage(t, fs, "/jobs/dequeue") + if second.Data != "second" { + t.Fatalf("second dequeue = %q, want second", second.Data) + } + + emptyPeek := mustReadAll(t, fs, "/jobs/peek") + if got := string(emptyPeek); got != "{}" { + t.Fatalf("empty peek = %q, want {}", got) + } + + emptyDequeue := mustReadAll(t, fs, "/jobs/dequeue") + if got := string(emptyDequeue); got != "{}" { + t.Fatalf("empty dequeue = %q, want {}", got) + } + + if _, err := fs.Write("/jobs/enqueue", []byte("to-clear"), -1, filesystem.WriteFlagAppend); err != nil { + t.Fatalf("enqueue before clear: %v", err) + } + if _, err := fs.Write("/jobs/clear", nil, -1, filesystem.WriteFlagTruncate); err != nil { + t.Fatalf("clear queue: %v", err) + } + + sizeData = mustReadAll(t, fs, "/jobs/size") + if got := string(sizeData); got != "0" { + t.Fatalf("queue size after clear = %q, want 0", got) + } +} + +func TestQueueFSPermissionsAndErrorsRegression(t *testing.T) { + fs := newTestQueueFS(t) + + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + + if _, err := fs.Read("/jobs/enqueue", 0, -1); err == nil || !strings.Contains(err.Error(), "write-only") { + t.Fatalf("read enqueue error = %v, want write-only", err) + } + if _, err := fs.Read("/jobs/clear", 0, -1); err == nil || !strings.Contains(err.Error(), "write-only") { + t.Fatalf("read clear error = %v, want write-only", err) + } + if _, err := fs.Write("/jobs/dequeue", []byte("x"), -1, filesystem.WriteFlagAppend); err == nil || !strings.Contains(err.Error(), "cannot write") { + t.Fatalf("write dequeue error = %v, want cannot write", err) + } + if _, err := fs.Write("/jobs/peek", []byte("x"), -1, filesystem.WriteFlagAppend); err == nil || !strings.Contains(err.Error(), "cannot write") { + t.Fatalf("write peek error = %v, want cannot write", err) + } + if _, err := fs.Write("/jobs/size", []byte("x"), -1, filesystem.WriteFlagAppend); err == nil || !strings.Contains(err.Error(), "cannot write") { + t.Fatalf("write size error = %v, want cannot write", err) + } + + if _, err := fs.Write("/jobs", []byte("x"), -1, filesystem.WriteFlagAppend); err == nil || !strings.Contains(err.Error(), "is a directory") { + t.Fatalf("write directory error = %v, want directory error", err) + } + if _, err := fs.Read("/enqueue", 0, -1); err == nil || !strings.Contains(err.Error(), "operation without queue name") { + t.Fatalf("read /enqueue error = %v, want missing queue name", err) + } + if _, err := fs.Stat("/jobs/unknown"); err == nil || !strings.Contains(err.Error(), "no such file or directory") { + t.Fatalf("stat unknown control path error = %v, want missing path", err) + } + if err := fs.Remove("/jobs"); err == nil || !strings.Contains(err.Error(), "use RemoveAll") { + t.Fatalf("remove directory error = %v, want RemoveAll guidance", err) + } + if err := fs.Remove("/jobs/enqueue"); err == nil || !strings.Contains(err.Error(), "cannot remove control files") { + t.Fatalf("remove control file error = %v, want control file error", err) + } + + entries, err := fs.ReadDir("/jobs") + if err != nil { + t.Fatalf("readdir /jobs: %v", err) + } + files := queueDirEntryNames(entries) + if got := files["enqueue"].Mode; got != 0o222 { + t.Fatalf("enqueue mode = %#o, want 0222", got) + } + if got := files["dequeue"].Mode; got != 0o444 { + t.Fatalf("dequeue mode = %#o, want 0444", got) + } + if got := files["size"].Meta.Type; got != MetaValueQueueStatus { + t.Fatalf("size meta type = %q, want %q", got, MetaValueQueueStatus) + } + if got := files["peek"].Meta.Type; got != MetaValueQueueControl { + t.Fatalf("peek meta type = %q, want %q", got, MetaValueQueueControl) + } +} diff --git a/agfs-server/pkg/plugins/queuefs/queuefs_handle_test.go b/agfs-server/pkg/plugins/queuefs/queuefs_handle_test.go new file mode 100644 index 0000000..0600669 --- /dev/null +++ b/agfs-server/pkg/plugins/queuefs/queuefs_handle_test.go @@ -0,0 +1,154 @@ +package queuefs + +import ( + "encoding/json" + "errors" + "io" + "strings" + "testing" + + "github.com/c4pt0r/agfs/agfs-server/pkg/filesystem" +) + +func TestQueueFSOpenWriteAndTruncateRegression(t *testing.T) { + fs := newTestQueueFS(t) + + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if err := fs.Truncate("/jobs/enqueue", 0); err != nil { + t.Fatalf("truncate enqueue: %v", err) + } + + writer, err := fs.OpenWrite("/jobs/enqueue") + if err != nil { + t.Fatalf("openwrite enqueue: %v", err) + } + if _, err := writer.Write([]byte("chunk-1")); err != nil { + t.Fatalf("write first chunk: %v", err) + } + if _, err := writer.Write([]byte("+chunk-2")); err != nil { + t.Fatalf("write second chunk: %v", err) + } + if err := writer.Close(); err != nil { + t.Fatalf("close enqueue writer: %v", err) + } + + msg := mustReadMessage(t, fs, "/jobs/dequeue") + if msg.Data != "chunk-1+chunk-2" { + t.Fatalf("openwrite enqueue payload = %q, want concatenated payload", msg.Data) + } + + if _, err := fs.Write("/jobs/enqueue", []byte("keep"), -1, filesystem.WriteFlagAppend); err != nil { + t.Fatalf("enqueue keep: %v", err) + } + clearWriter, err := fs.OpenWrite("/jobs/clear") + if err != nil { + t.Fatalf("openwrite clear: %v", err) + } + if err := clearWriter.Close(); err != nil { + t.Fatalf("close clear writer: %v", err) + } + + sizeData := mustReadAll(t, fs, "/jobs/size") + if got := string(sizeData); got != "0" { + t.Fatalf("queue size after openwrite clear = %q, want 0", got) + } +} + +func TestQueueFSHandleRegression(t *testing.T) { + fs := newTestQueueFS(t) + + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + + enqueueHandle, err := fs.OpenHandle("/jobs/enqueue", filesystem.O_WRONLY, 0) + if err != nil { + t.Fatalf("open enqueue handle: %v", err) + } + if _, err := enqueueHandle.Write([]byte("payload")); err != nil { + t.Fatalf("write enqueue handle: %v", err) + } + + stat, err := enqueueHandle.Stat() + if err != nil { + t.Fatalf("stat enqueue handle: %v", err) + } + if stat.Mode != 0o222 { + t.Fatalf("enqueue handle mode = %#o, want 0222", stat.Mode) + } + if enqueueHandle.Flags() != filesystem.O_WRONLY { + t.Fatalf("enqueue handle flags = %v, want O_WRONLY", enqueueHandle.Flags()) + } + + sizeHandle, err := fs.OpenHandle("/jobs/size", filesystem.O_RDONLY, 0) + if err != nil { + t.Fatalf("open size handle: %v", err) + } + buf := make([]byte, 4) + n, err := sizeHandle.Read(buf) + if err != nil && !errors.Is(err, io.EOF) { + t.Fatalf("read size handle: %v", err) + } + if got := string(buf[:n]); got != "1" { + t.Fatalf("size handle read = %q, want 1", got) + } + if gotHandle, err := fs.GetHandle(sizeHandle.ID()); err != nil || gotHandle.ID() != sizeHandle.ID() { + t.Fatalf("get size handle = (%v, %v), want same handle", gotHandle, err) + } + + dequeueHandle, err := fs.OpenHandle("/jobs/dequeue", filesystem.O_RDONLY, 0) + if err != nil { + t.Fatalf("open dequeue handle: %v", err) + } + + firstChunk := make([]byte, 8) + n, err = dequeueHandle.Read(firstChunk) + if err != nil && !errors.Is(err, io.EOF) { + t.Fatalf("read dequeue first chunk: %v", err) + } + secondChunk := make([]byte, 256) + n2, err := dequeueHandle.Read(secondChunk) + if err != nil && !errors.Is(err, io.EOF) { + t.Fatalf("read dequeue second chunk: %v", err) + } + if _, err := dequeueHandle.Read(secondChunk); !errors.Is(err, io.EOF) { + t.Fatalf("expected EOF after cached dequeue payload, got %v", err) + } + + payload := append(firstChunk[:n], secondChunk[:n2]...) + var msg QueueMessage + if err := json.Unmarshal(payload, &msg); err != nil { + t.Fatalf("unmarshal dequeue payload: %v (payload=%q)", err, string(payload)) + } + if msg.Data != "payload" { + t.Fatalf("dequeue handle payload = %q, want payload", msg.Data) + } + + sizeData := mustReadAll(t, fs, "/jobs/size") + if got := string(sizeData); got != "0" { + t.Fatalf("queue size after dequeue handle = %q, want 0", got) + } + + if _, err := sizeHandle.Write([]byte("x")); err == nil || !strings.Contains(err.Error(), "cannot write") { + t.Fatalf("write size handle error = %v, want cannot write", err) + } + + if err := dequeueHandle.Close(); err != nil { + t.Fatalf("close dequeue handle: %v", err) + } + if err := fs.CloseHandle(sizeHandle.ID()); err != nil { + t.Fatalf("close size handle by id: %v", err) + } + if _, err := fs.GetHandle(sizeHandle.ID()); !errors.Is(err, filesystem.ErrNotFound) { + t.Fatalf("get closed size handle error = %v, want ErrNotFound", err) + } + + if err := enqueueHandle.Close(); err != nil { + t.Fatalf("close enqueue handle: %v", err) + } + if err := fs.CloseHandle(enqueueHandle.ID()); !errors.Is(err, filesystem.ErrNotFound) { + t.Fatalf("close already closed enqueue handle error = %v, want ErrNotFound", err) + } +} diff --git a/agfs-server/pkg/plugins/queuefs/sqlite_backend_test.go b/agfs-server/pkg/plugins/queuefs/sqlite_backend_test.go new file mode 100644 index 0000000..d2365db --- /dev/null +++ b/agfs-server/pkg/plugins/queuefs/sqlite_backend_test.go @@ -0,0 +1,142 @@ +package queuefs + +import ( + "path/filepath" + "testing" +) + +func newSQLiteTestQueueFS(t *testing.T, dbPath string) *queueFS { + t.Helper() + + plugin := NewQueueFSPlugin() + if err := plugin.Initialize(map[string]interface{}{ + "backend": "sqlite", + "db_path": dbPath, + }); err != nil { + t.Fatalf("initialize sqlite queuefs: %v", err) + } + t.Cleanup(func() { + if plugin.backend != nil { + _ = plugin.backend.Close() + } + }) + + fs, ok := plugin.GetFileSystem().(*queueFS) + if !ok { + t.Fatalf("unexpected filesystem type %T", plugin.GetFileSystem()) + } + return fs +} + +func TestQueueFSSQLiteFileRegression(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "queuefs.db") + fs := newSQLiteTestQueueFS(t, dbPath) + + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if err := fs.Mkdir("/logs/errors", 0o755); err != nil { + t.Fatalf("mkdir /logs/errors: %v", err) + } + + entries, err := fs.ReadDir("/") + if err != nil { + t.Fatalf("readdir root: %v", err) + } + rootEntries := queueDirEntryNames(entries) + for _, name := range []string{"README", "jobs", "logs"} { + if _, ok := rootEntries[name]; !ok { + t.Fatalf("root missing %q in %+v", name, entries) + } + } + + if _, err := fs.Write("/jobs/enqueue", []byte("first"), -1, 0); err != nil { + t.Fatalf("enqueue first: %v", err) + } + if _, err := fs.Write("/jobs/enqueue", []byte("second"), -1, 0); err != nil { + t.Fatalf("enqueue second: %v", err) + } + + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "2" { + t.Fatalf("queue size = %q, want 2", got) + } + + peeked := mustReadMessage(t, fs, "/jobs/peek") + if peeked.Data != "first" { + t.Fatalf("peeked message = %q, want first", peeked.Data) + } + + first := mustReadMessage(t, fs, "/jobs/dequeue") + second := mustReadMessage(t, fs, "/jobs/dequeue") + if first.Data != "first" || second.Data != "second" { + t.Fatalf("dequeue order = [%q, %q], want [first, second]", first.Data, second.Data) + } + + if got := string(mustReadAll(t, fs, "/jobs/dequeue")); got != "{}" { + t.Fatalf("empty dequeue = %q, want {}", got) + } + + if _, err := fs.Write("/jobs/enqueue", []byte("to-clear"), -1, 0); err != nil { + t.Fatalf("enqueue before clear: %v", err) + } + if _, err := fs.Write("/jobs/clear", nil, -1, 0); err != nil { + t.Fatalf("clear queue: %v", err) + } + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "0" { + t.Fatalf("queue size after clear = %q, want 0", got) + } + + if err := fs.RemoveAll("/logs"); err != nil { + t.Fatalf("removeall /logs: %v", err) + } + if _, err := fs.Stat("/logs/errors"); err == nil { + t.Fatal("expected removed nested sqlite queue to disappear") + } +} + +func TestQueueFSSQLitePersistenceRegression(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "queuefs-persist.db") + + func() { + fs := newSQLiteTestQueueFS(t, dbPath) + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if _, err := fs.Write("/jobs/enqueue", []byte("persisted"), -1, 0); err != nil { + t.Fatalf("enqueue persisted message: %v", err) + } + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "1" { + t.Fatalf("initial queue size = %q, want 1", got) + } + }() + + fs := newSQLiteTestQueueFS(t, dbPath) + + entries, err := fs.ReadDir("/") + if err != nil { + t.Fatalf("readdir root after reopen: %v", err) + } + if _, ok := queueDirEntryNames(entries)["jobs"]; !ok { + t.Fatalf("root missing reopened queue in %+v", entries) + } + + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "1" { + t.Fatalf("reopened queue size = %q, want 1", got) + } + peeked := mustReadMessage(t, fs, "/jobs/peek") + if peeked.Data != "persisted" { + t.Fatalf("peek after reopen = %q, want persisted", peeked.Data) + } + + dequeued := mustReadMessage(t, fs, "/jobs/dequeue") + if dequeued.Data != "persisted" { + t.Fatalf("dequeue after reopen = %q, want persisted", dequeued.Data) + } + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "0" { + t.Fatalf("queue size after reopened dequeue = %q, want 0", got) + } + + if _, err := fs.Stat("/jobs"); err != nil { + t.Fatalf("stat empty queue after reopen: %v", err) + } +} diff --git a/agfs-server/pkg/plugins/queuefs/tidb_backend_test.go b/agfs-server/pkg/plugins/queuefs/tidb_backend_test.go new file mode 100644 index 0000000..14e2c0c --- /dev/null +++ b/agfs-server/pkg/plugins/queuefs/tidb_backend_test.go @@ -0,0 +1,253 @@ +package queuefs + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "os" + "sync" + "testing" + "time" + + "github.com/go-sql-driver/mysql" +) + +func tidbTestConfig(t *testing.T, database string) map[string]interface{} { + t.Helper() + + if os.Getenv("TIDB_TEST") == "" { + t.Skip("set TIDB_TEST=1 to run TiDB integration tests") + } + + dsn := os.Getenv("TIDB_TEST_DSN") + if dsn == "" { + t.Skip("set TIDB_TEST_DSN to run TiDB integration tests") + } + + parsedDSN, err := mysql.ParseDSN(dsn) + if err != nil { + t.Fatalf("parse TIDB_TEST_DSN: %v", err) + } + parsedDSN.DBName = database + if parsedDSN.Params == nil { + parsedDSN.Params = map[string]string{} + } + + return map[string]interface{}{ + "backend": "tidb", + "dsn": parsedDSN.FormatDSN(), + "database": database, + } +} + +func newTiDBTestQueueFS(t *testing.T, database string) *queueFS { + t.Helper() + + plugin := NewQueueFSPlugin() + if err := plugin.Initialize(tidbTestConfig(t, database)); err != nil { + t.Fatalf("initialize tidb queuefs: %v", err) + } + t.Cleanup(func() { + if plugin.backend != nil { + _ = plugin.backend.Close() + } + }) + + fs, ok := plugin.GetFileSystem().(*queueFS) + if !ok { + t.Fatalf("unexpected filesystem type %T", plugin.GetFileSystem()) + } + return fs +} + +func newTiDBTestDatabaseName(t *testing.T) string { + t.Helper() + + return fmt.Sprintf("queuefs_test_%d", time.Now().UnixNano()) +} + +func TestQueueFSTiDBFileRegression(t *testing.T) { + database := newTiDBTestDatabaseName(t) + fs := newTiDBTestQueueFS(t, database) + + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if err := fs.Mkdir("/logs/errors", 0o755); err != nil { + t.Fatalf("mkdir /logs/errors: %v", err) + } + + entries, err := fs.ReadDir("/") + if err != nil { + t.Fatalf("readdir root: %v", err) + } + rootEntries := queueDirEntryNames(entries) + for _, name := range []string{"README", "jobs", "logs"} { + if _, ok := rootEntries[name]; !ok { + t.Fatalf("root missing %q in %+v", name, entries) + } + } + + if _, err := fs.Write("/jobs/enqueue", []byte("first"), -1, 0); err != nil { + t.Fatalf("enqueue first: %v", err) + } + if _, err := fs.Write("/jobs/enqueue", []byte("second"), -1, 0); err != nil { + t.Fatalf("enqueue second: %v", err) + } + + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "2" { + t.Fatalf("queue size = %q, want 2", got) + } + + peeked := mustReadMessage(t, fs, "/jobs/peek") + if peeked.Data != "first" { + t.Fatalf("peeked message = %q, want first", peeked.Data) + } + + first := mustReadMessage(t, fs, "/jobs/dequeue") + second := mustReadMessage(t, fs, "/jobs/dequeue") + if first.Data != "first" || second.Data != "second" { + t.Fatalf("dequeue order = [%q, %q], want [first, second]", first.Data, second.Data) + } + + if got := string(mustReadAll(t, fs, "/jobs/dequeue")); got != "{}" { + t.Fatalf("empty dequeue = %q, want {}", got) + } + + if _, err := fs.Write("/jobs/enqueue", []byte("to-clear"), -1, 0); err != nil { + t.Fatalf("enqueue before clear: %v", err) + } + if _, err := fs.Write("/jobs/clear", nil, -1, 0); err != nil { + t.Fatalf("clear queue: %v", err) + } + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "0" { + t.Fatalf("queue size after clear = %q, want 0", got) + } + + if err := fs.RemoveAll("/logs"); err != nil { + t.Fatalf("removeall /logs: %v", err) + } + if _, err := fs.Stat("/logs/errors"); err == nil { + t.Fatal("expected removed nested tidb queue to disappear") + } +} + +func TestQueueFSTiDBPersistenceRegression(t *testing.T) { + database := newTiDBTestDatabaseName(t) + + func() { + fs := newTiDBTestQueueFS(t, database) + if err := fs.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if _, err := fs.Write("/jobs/enqueue", []byte("persisted"), -1, 0); err != nil { + t.Fatalf("enqueue persisted message: %v", err) + } + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "1" { + t.Fatalf("initial queue size = %q, want 1", got) + } + }() + + fs := newTiDBTestQueueFS(t, database) + + entries, err := fs.ReadDir("/") + if err != nil { + t.Fatalf("readdir root after reopen: %v", err) + } + if _, ok := queueDirEntryNames(entries)["jobs"]; !ok { + t.Fatalf("root missing reopened queue in %+v", entries) + } + + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "1" { + t.Fatalf("reopened queue size = %q, want 1", got) + } + peeked := mustReadMessage(t, fs, "/jobs/peek") + if peeked.Data != "persisted" { + t.Fatalf("peek after reopen = %q, want persisted", peeked.Data) + } + + dequeued := mustReadMessage(t, fs, "/jobs/dequeue") + if dequeued.Data != "persisted" { + t.Fatalf("dequeue after reopen = %q, want persisted", dequeued.Data) + } + if got := string(mustReadAll(t, fs, "/jobs/size")); got != "0" { + t.Fatalf("queue size after reopened dequeue = %q, want 0", got) + } + + if _, err := fs.Stat("/jobs"); err != nil { + t.Fatalf("stat empty queue after reopen: %v", err) + } +} + +func TestQueueFSTiDBConfigUsesDSN(t *testing.T) { + if os.Getenv("TIDB_TEST") == "" { + t.Skip("set TIDB_TEST=1 to run TiDB integration tests") + } + + config := tidbTestConfig(t, newTiDBTestDatabaseName(t)) + if _, ok := config["dsn"].(string); !ok || config["dsn"] == "" { + t.Fatalf("expected non-empty dsn in config: %+v", config) + } +} + +func TestQueueFSTiDBConcurrentDequeueRegression(t *testing.T) { + database := newTiDBTestDatabaseName(t) + writerFS := newTiDBTestQueueFS(t, database) + readerOne := newTiDBTestQueueFS(t, database) + readerTwo := newTiDBTestQueueFS(t, database) + + if err := writerFS.Mkdir("/jobs", 0o755); err != nil { + t.Fatalf("mkdir /jobs: %v", err) + } + if _, err := writerFS.Write("/jobs/enqueue", []byte("once"), -1, 0); err != nil { + t.Fatalf("enqueue once: %v", err) + } + + type dequeueResult struct { + payload []byte + err error + } + + start := make(chan struct{}) + results := make(chan dequeueResult, 2) + var wg sync.WaitGroup + for _, fs := range []*queueFS{readerOne, readerTwo} { + wg.Add(1) + go func(fs *queueFS) { + defer wg.Done() + <-start + payload, err := fs.Read("/jobs/dequeue", 0, -1) + results <- dequeueResult{payload: payload, err: err} + }(fs) + } + close(start) + wg.Wait() + close(results) + + nonEmpty := 0 + for result := range results { + if result.err != nil && !errors.Is(result.err, io.EOF) { + t.Fatalf("concurrent dequeue: %v", result.err) + } + if string(result.payload) == "{}" { + continue + } + + var msg QueueMessage + if err := json.Unmarshal(result.payload, &msg); err != nil { + t.Fatalf("unmarshal concurrent dequeue payload: %v (payload=%q)", err, string(result.payload)) + } + if msg.Data != "once" { + t.Fatalf("concurrent dequeue returned %q, want once", msg.Data) + } + nonEmpty++ + } + + if nonEmpty != 1 { + t.Fatalf("concurrent dequeue claimed %d messages, want 1", nonEmpty) + } + if got := string(mustReadAll(t, writerFS, "/jobs/size")); got != "0" { + t.Fatalf("queue size after concurrent dequeue = %q, want 0", got) + } +}