abstract: fix switch in_replicaset value

.github/workflows/publish.yml

@@ -24,7 +24,7 @@ jobs:
 
       # Create a rockspec for the release.
       - run: printf '%s=%s\n' TAG "${GITHUB_REF##*/}" >> "${GITHUB_ENV}"
-      sed -E
+      - run: sed -E
           -e "s/branch = '.+'/tag = '${{ env.TAG }}'/g"
           -e "s/version = '.+'/version = '${{ env.TAG }}-1'/g"
           queue-scm-1.rockspec > queue-${{ env.TAG }}-1.rockspec

README.md

@@ -371,7 +371,7 @@ flowchart LR
       I(("init"))-->S[startup]
       S[startup]-->R[running]
       W[waiting]--> |"(ro ->rw)"| S[startup]
-      R[running]--> |"(ro ->rw)"| E[ending]
+      R[running]--> |"(rw ->ro)"| E[ending]
       E[ending]-->W[waiting]
 ```
 

queue/abstract.lua

@@ -11,6 +11,10 @@ local qc       = require('queue.compat')
 local num_type = qc.num_type
 local str_type = qc.str_type
 
+-- since:
+-- https://github.com/tarantool/tarantool/commit/f266559b167b4643c377724eebde9651877d6cc9
+local ddl_txn_supported = qc.check_version({2, 2, 1})
+
 -- The term "queue session" has been added to the queue. One "queue session"
 -- can include many connections (box.session). For clarity, the box.session
 -- will be referred to as connection below.
@@ -579,8 +583,7 @@ function method.create_tube(tube_name, tube_type, opts)
 
     local replicaset_mode = queue.cfg['in_replicaset'] or false
     if replicaset_mode and opts.temporary then
-        log.error("Cannot create temporary tube in replicaset mode")
-        return
+        error("Cannot create temporary tube in replicaset mode")
     end
 
     local driver = queue.driver[tube_type]
@@ -669,13 +672,23 @@ local function switch_in_replicaset(replicaset_mode)
         box.space._queue_taken_2_mgr:insert(tuple)
     end
 
-    box.space._queue_taken_2:drop()
-    box.space._queue_taken_2_mgr:rename('_queue_taken_2')
+    if ddl_txn_supported then
+		-- We can do it inside a transaction.
+        box.space._queue_taken_2:drop()
+        box.space._queue_taken_2_mgr:rename('_queue_taken_2')
+    end
 
     local status, err = pcall(box.commit)
     if not status then
         error(('Error migrate _queue_taken_2: %s'):format(tostring(err)))
     end
+
+    if not ddl_txn_supported then
+		-- Do it outside a transaction becase DDL does not support
+		-- multi-statement transactions.
+        box.space._queue_taken_2:drop()
+        box.space._queue_taken_2_mgr:rename('_queue_taken_2')
+    end
 end
 
 -- create or join infrastructure

queue/abstract/queue_session.lua

@@ -7,6 +7,9 @@ local qc       = require('queue.compat')
 local num_type = qc.num_type
 local str_type = qc.str_type
 
+-- since:
+-- https://github.com/tarantool/tarantool/commit/f266559b167b4643c377724eebde9651877d6cc9
+local ddl_txn_supported = qc.check_version({2, 2, 1})
 
 local queue_session = {}
 
@@ -62,13 +65,23 @@ local function switch_in_replicaset(replicaset_mode)
         box.space._queue_shared_sessions_mgr:insert(tuple)
     end
 
-    box.space._queue_shared_sessions:drop()
-    box.space._queue_shared_sessions_mgr:rename('_queue_shared_sessions')
+    if ddl_txn_supported then
+		-- We can do it inside a transaction.
+        box.space._queue_shared_sessions:drop()
+        box.space._queue_shared_sessions_mgr:rename('_queue_shared_sessions')
+    end
 
     local status, err = pcall(box.commit)
     if not status then
         error(('Error migrate _queue_shared_sessions: %s'):format(tostring(err)))
     end
+
+    if not ddl_txn_supported then
+		-- Do it outside a transaction becase DDL does not support
+		-- multi-statement transactions.
+        box.space._queue_shared_sessions:drop()
+        box.space._queue_shared_sessions_mgr:rename('_queue_shared_sessions')
+    end
 end
 
 --- Create everything that's needed to work with "shared" sessions.

t/190-work-with-ttl-buried-task.t

@@ -52,7 +52,7 @@ test:test('test work with "ttl", when "bury" after "take"', function(test)
 
         -- Check status of the task after "ttl" has expired.
         fiber.sleep(TTL * 2)
-        ok, res = pcall(tube.peek, tube, id)
+        local ok, res = pcall(tube.peek, tube, id)
         test:ok(res:match(string.format('Task %d not found', id)),
             ('task done, driver: "%s"'):format(driver))
 

t/210-cfg-in-replicaset.t

@@ -0,0 +1,43 @@
+#!/usr/bin/env tarantool
+
+local test  = require('tap').test('')
+local queue = require('queue')
+
+rawset(_G, 'queue', require('queue'))
+
+test:plan(2)
+
+test:test('Check in_replicaset switches', function(test)
+    test:plan(4)
+    local engine = os.getenv('ENGINE') or 'memtx'
+
+    box.cfg({})
+
+    local status, err = pcall(queue.cfg, {in_replicaset = true})
+    test:ok(status, 'in_replicaset = true switched')
+    local status, _ = pcall(queue.cfg, {in_replicaset = false})
+    test:ok(status, 'in_replicaset = false switched')
+    local status, _ = pcall(queue.cfg, {in_replicaset = true})
+    test:ok(status, 'in_replicaset = true switched')
+    local status, _ = pcall(queue.cfg, {in_replicaset = false})
+    test:ok(status, 'in_replicaset = false switched')
+end)
+
+test:test('Check error create temporary tube', function(test)
+    test:plan(2)
+    local engine = os.getenv('ENGINE') or 'memtx'
+
+    box.cfg({})
+
+    queue.cfg{ttr = 0.5, in_replicaset = true}
+    local opts = {temporary = true, engine = engine}
+    local status, err = pcall(queue.create_tube, 'test', 'fifo', opts)
+    test:is(status, false, 'test tube should not be created')
+    local founded = string.find(err,
+        'Cannot create temporary tube in replicaset mode')
+    test:ok(founded, 'unexpected error')
+end)
+
+rawset(_G, 'queue', nil)
+os.exit(test:check() and 0 or 1)
+-- vim: set ft=lua :