Class: Tina4::Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/tina4/queue.rb

Overview

Queue — unified wrapper for queue management operations. Auto-detects backend from TINA4_QUEUE_BACKEND env var.

Usage:

# Auto-detect from env (default: lite/file backend)
queue = Queue.new(topic: "tasks")

# Explicit backend
queue = Queue.new(topic: "tasks", backend: :rabbitmq)

# Or pass a backend instance directly (legacy)
queue = Queue.new(topic: "tasks", backend: my_backend)

Constant Summary collapse

JOB_EXTENSION =

Canonical job-file extension in all four frameworks.

".queue-data"
DEAD_LETTER_DIRNAME =

Dead letters share one directory and carry their topic in the record.

"dead_letter"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(topic:, backend: nil, max_retries: 3, retry_backoff: 0, visibility_timeout: nil) ⇒ Queue

Returns a new instance of Queue.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/tina4/queue.rb', line 23

def initialize(topic:, backend: nil, max_retries: 3, retry_backoff: 0, visibility_timeout: nil)
  @topic = topic
  @max_retries = max_retries
  # Seconds to wait before a failed job is re-attempted (lite backend).
  # Default 0 = retry on the very next pop/consume iteration.
  @retry_backoff = retry_backoff
  # Reservation/visibility timeout (seconds). A popped job is reserved for
  # this long; if the consumer dies before complete()/fail() the next pop()
  # reclaims it (at-least-once delivery). Falls back to
  # TINA4_QUEUE_VISIBILITY_TIMEOUT, else 300 (5 min). <= 0 disables reclaim.
  # RabbitMQ/Kafka ignore it — the broker owns redelivery.
  @visibility_timeout =
    visibility_timeout.nil? ? self.class.default_visibility_timeout : visibility_timeout.to_f
  @backend = resolve_backend_arg(backend)
end

Instance Attribute Details

#max_retries ⇒ Object (readonly)

Returns the value of attribute max_retries.



21
22
23
 
# File 'lib/tina4/queue.rb', line 21

def max_retries
  @max_retries
end

#retry_backoff ⇒ Object (readonly)

Returns the value of attribute retry_backoff.



21
22
23
 
# File 'lib/tina4/queue.rb', line 21

def retry_backoff
  @retry_backoff
end

#topic ⇒ Object (readonly)

Returns the value of attribute topic.



21
22
23
 
# File 'lib/tina4/queue.rb', line 21

def topic
  @topic
end

#visibility_timeout ⇒ Object (readonly)

Returns the value of attribute visibility_timeout.



21
22
23
 
# File 'lib/tina4/queue.rb', line 21

def visibility_timeout
  @visibility_timeout
end

Class Method Details

.base_path ⇒ Object

Root of the file store: TINA4_QUEUE_PATH, else "data/queue" relative to the working directory.

Absolute, so every caller gets the same directory rather than one that depends on when it was resolved. A blank value is treated as unset (the same normalisation TINA4_SESSION_BACKEND applies) — File.join("", topic) would otherwise silently address /, at the filesystem root.



70
71
72
73
 
# File 'lib/tina4/queue.rb', line 70

def self.base_path
  configured = ENV["TINA4_QUEUE_PATH"].to_s.strip
  File.expand_path(configured.empty? ? "data/queue" : configured)
end

.default_visibility_timeout ⇒ Object

Reservation/visibility timeout in seconds, from env (default 300 = 5 min).



101
102
103
104
105
 
# File 'lib/tina4/queue.rb', line 101

def self.default_visibility_timeout
  Float(ENV.fetch("TINA4_QUEUE_VISIBILITY_TIMEOUT", "300"))
rescue ArgumentError, TypeError
  300.0
end

.job_file(dir, id) ⇒ Object 



96
97
98
 
# File 'lib/tina4/queue.rb', line 96

def self.job_file(dir, id)
  File.join(dir, "#{id}#{JOB_EXTENSION}")
end

.job_files(dir) ⇒ Object

Every job file in dir, name-sorted so two readers of the same directory always see the same order.



92
93
94
 
# File 'lib/tina4/queue.rb', line 92

def self.job_files(dir)
  Dir.glob(File.join(dir, "*#{JOB_EXTENSION}")).sort
end

.parse_amqp_url(url) ⇒ Object 



516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
 
# File 'lib/tina4/queue.rb', line 516

def self.parse_amqp_url(url)
  config = {}
  url = url.sub("amqp://", "").sub("amqps://", "")

  if url.include?("@")
    creds, rest = url.split("@", 2)
    if creds.include?(":")
      config[:username], config[:password] = creds.split(":", 2)
    else
      config[:username] = creds
    end
  else
    rest = url
  end

  if rest.include?("/")
    hostport, vhost = rest.split("/", 2)
    # THE VHOST IS THE PATH SEGMENT, URL-DECODED, WITH NO LEADING SLASH
    # (RabbitMQ URI spec). This used to prepend "/", so
    # amqp://guest:guest@rabbit:5672/orders asked for a vhost literally
    # named "/orders". No broker has that one - it is named "orders" - so
    # every publish failed against a named vhost, which is the ordinary
    # multi-tenant setup and the form every RabbitMQ tutorial shows.
    # MEASURED against a real broker: 4 of 5 URL shapes resolved to the
    # wrong name, and the only one that worked carried no vhost at all,
    # which is why four green suites never noticed.
    #
    # Decoding matters for the same reason: the DEFAULT vhost is named "/",
    # which cannot appear literally in a path, so the spec spells it "%2f".
    #
    # DELIBERATE DEVIATION, one shape: the spec reads a bare trailing slash
    # as the EMPTY vhost name. Tina4 treats it as "not specified" and keeps
    # the caller's default - nobody writes a trailing slash intending a
    # vhost named "", and reading it literally would break a working
    # "amqp://host:5672/" for no benefit.
    config[:vhost] = URI::DEFAULT_PARSER.unescape(vhost) if vhost && !vhost.empty?
  else
    hostport = rest
  end

  if hostport.include?(":")
    host, port = hostport.split(":", 2)
    config[:host] = host
    config[:port] = port.to_i
  elsif hostport && !hostport.empty?
    config[:host] = hostport
  end

  config
end

.resolve_backend(name = nil, max_retries: 3, retry_backoff: 0, visibility_timeout: nil) ⇒ Object

Resolve the default backend from env vars.



406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
 
# File 'lib/tina4/queue.rb', line 406

def self.resolve_backend(name = nil, max_retries: 3, retry_backoff: 0, visibility_timeout: nil)
  # Normalise BOTH sources, not just the env var. `.downcase.strip` used to
  # bind only to the ENV.fetch branch, so an explicit
  # Queue.new(backend: "FILE") fell through to the unknown-backend raise
  # while the identical spelling in TINA4_QUEUE_BACKEND resolved fine -
  # and while python, php and nodejs all accepted it. Python is master here.
  chosen = (name || ENV.fetch("TINA4_QUEUE_BACKEND", "file")).to_s.downcase.strip
  vt = visibility_timeout.nil? ? default_visibility_timeout : visibility_timeout

  case chosen.to_s
  when "lite", "file", "default"
    Tina4::QueueBackends::LiteBackend.new(
      max_retries: max_retries, retry_backoff: retry_backoff, visibility_timeout: vt
    )
  when "rabbitmq"
    # Broker manages visibility/redelivery (unacked messages requeue on
    # channel close) — the framework timeout is accepted but not used.
    # max_retries IS used: fail() counts attempts itself and dead-letters
    # past the limit. Without threading it through, the backend fell back to
    # its own default of 3, so Queue.new(max_retries: 2) gave a job THREE
    # attempts on rabbitmq and two on file — the dead-letter threshold
    # silently changed with the provider.
    config = resolve_rabbitmq_config
    config[:max_retries] = max_retries
    Tina4::QueueBackends::RabbitmqBackend.new(config)
  when "kafka"
    # Consumer-group offsets manage redelivery — framework timeout N/A.
    # max_retries threaded through for the same reason as rabbitmq.
    config = resolve_kafka_config
    config[:max_retries] = max_retries
    Tina4::QueueBackends::KafkaBackend.new(config)
  when "mongodb", "mongo"
    config = resolve_mongo_config
    config[:visibility_timeout] = vt
    config[:max_retries] = max_retries
    # Thread retry_backoff through so a failed/retried job's available_at is
    # reset to now (or now + retry_backoff) instead of being stranded for the
    # full visibility window (Bug B).
    config[:retry_backoff] = retry_backoff
    Tina4::QueueBackends::MongoBackend.new(config)
  else
    raise ArgumentError, "Unknown queue backend: #{chosen.inspect}. Use 'lite', 'rabbitmq', 'kafka', or 'mongodb'."
  end
end

.resolve_kafka_config ⇒ Object 



488
489
490
491
492
493
494
495
496
497
498
499
 
# File 'lib/tina4/queue.rb', line 488

def self.resolve_kafka_config
  config = {}
  url = ENV["TINA4_QUEUE_URL"]
  if url
    config[:brokers] = url.sub("kafka://", "")
  end
  brokers = ENV["TINA4_KAFKA_BROKERS"]
  config[:brokers] = brokers if brokers
  config[:brokers] ||= "localhost:9092"
  config[:group_id] = ENV.fetch("TINA4_KAFKA_GROUP_ID", "tina4_consumer_group")
  config
end

.resolve_mongo_config ⇒ Object 



501
502
503
504
505
506
507
508
509
510
511
512
513
514
 
# File 'lib/tina4/queue.rb', line 501

def self.resolve_mongo_config
  config = {}
  uri = ENV["TINA4_MONGO_URI"]
  config[:uri] = uri if uri
  config[:host] = ENV.fetch("TINA4_MONGO_HOST", "localhost") unless uri
  config[:port] = (ENV["TINA4_MONGO_PORT"] || 27017).to_i unless uri
  username = ENV["TINA4_MONGO_USERNAME"]
  password = ENV["TINA4_MONGO_PASSWORD"]
  config[:username] = username if username
  config[:password] = password if password
  config[:db] = ENV.fetch("TINA4_MONGO_DB", "tina4")
  config[:collection] = ENV.fetch("TINA4_MONGO_COLLECTION", "tina4_queue")
  config
end

.resolve_rabbitmq_config ⇒ Object 



474
475
476
477
478
479
480
481
482
483
484
485
486
 
# File 'lib/tina4/queue.rb', line 474

def self.resolve_rabbitmq_config
  config = {}
  url = ENV["TINA4_QUEUE_URL"]
  if url
    config = parse_amqp_url(url)
  end
  config[:host] ||= ENV.fetch("TINA4_RABBITMQ_HOST", "localhost")
  config[:port] ||= (ENV["TINA4_RABBITMQ_PORT"] || 5672).to_i
  config[:username] ||= ENV.fetch("TINA4_RABBITMQ_USERNAME", "guest")
  config[:password] ||= ENV.fetch("TINA4_RABBITMQ_PASSWORD", "guest")
  config[:vhost] ||= ENV.fetch("TINA4_RABBITMQ_VHOST", "/")
  config
end

.topic_dirname(topic) ⇒ Object

The single path segment a topic maps to.

Sanitised HERE, in the one place that resolves it, so no caller can build a store path out of an unsanitised topic name — the dev-admin queue panel takes its topic straight off a query string, where "../.." would otherwise walk out of the store.



81
82
83
 
# File 'lib/tina4/queue.rb', line 81

def self.topic_dirname(topic)
  topic.to_s.gsub(/[^a-zA-Z0-9_-]/, "_")
end

.topic_path(topic) ⇒ Object

Directory holding +topic+'s pending jobs.



86
87
88
 
# File 'lib/tina4/queue.rb', line 86

def self.topic_path(topic)
  File.join(base_path, topic_dirname(topic))
end

Instance Method Details

#backend ⇒ Object

Get the underlying backend instance.



401
402
403
 
# File 'lib/tina4/queue.rb', line 401

def backend
  @backend
end

#clear ⇒ Object

Clear all pending jobs from this queue's topic. Returns count removed.



141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/tina4/queue.rb', line 141

def clear # -> int
  # Was `return 0` - a SILENT no-op that looked exactly like an empty
  # queue. A backend that cannot clear says so, naming itself.
  unless @backend.respond_to?(:clear)
    raise NotImplementedError,
          "The #{@backend.class.name.split('::').last} queue backend cannot " \
          "perform clear(): it cannot remove jobs by topic. Use the file or " \
          "mongodb backend."
  end
  @backend.clear(@topic)
end

#close ⇒ Object

Release the backend's connection and free its resources.

A queue on RabbitMQ, Kafka or MongoDB holds a REAL client and socket. Until 3.13.95 there was no way to hand it back: close existed on the rabbitmq/mongo/kafka backends but was surfaced on NOTHING, and the lite backend had none at all - so every backend.close if respond_to?(:close) guard in the tree silently skipped the default backend. An app that built a Queue per request leaked one client per request, invisibly, until the broker refused new connections. Same class of leak as ADR-0025 corollary 4 (client-lifecycle-is-bounded).

Safe on EVERY backend: the lite backend holds no connection and closes as a documented no-op, so a TINA4_QUEUE_BACKEND change never turns a working shutdown path into an error. Idempotent - each backend drops its handles on the first call, so a second call finds nothing to close and returns.

Treat the queue as spent afterwards and build a new one to keep working.



170
171
172
 
# File 'lib/tina4/queue.rb', line 170

def close
  @backend.close
end

#consume(topic = nil, id: nil, poll_interval: 1.0, iterations: 0, batch_size: 1, &block) ⇒ Object

Consume jobs from a topic using an Enumerator (yield pattern).

Usage:

queue.consume("emails") do |job|
process(job)
end

# Consume a specific job by ID:
queue.consume("emails", id: "abc-123") do |job|
process(job)
end

# Or as an enumerator:
queue.consume("emails").each { |job| process(job) }

Consume jobs from a topic using a long-running generator.

Polls the queue continuously. When empty, sleeps for poll_interval seconds before polling again. No external while-loop or sleep needed.

queue.consume("emails") { |job| process(job) }
queue.consume("emails", poll_interval: 5) { |job| process(job) }
queue.consume("emails", id: "abc-123") { |job| process(job) }


257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
 
# File 'lib/tina4/queue.rb', line 257

def consume(topic = nil, id: nil, poll_interval: 1.0, iterations: 0, batch_size: 1, &block)
  topic ||= @topic

  if id
    # Single job by ID — no polling
    job = pop_by_id(topic, id)
    if job
      block_given? ? yield(job) : (return Enumerator.new { |y| y << job })
    end
    return block_given? ? nil : Enumerator.new { |_| }
  end

  # poll_interval=0 → single-pass drain (returns when empty)
  # poll_interval>0 → long-running poll (sleeps when empty, never returns)
  # iterations>0    → stop after consuming N jobs
  if block_given?
    consumed = 0
    if batch_size > 1
      loop do
        jobs = pop_batch(batch_size)
        if jobs.empty?
          break if poll_interval <= 0
          sleep(poll_interval)
          next
        end
        yield jobs
        consumed += jobs.length
        break if iterations > 0 && consumed >= iterations
      end
    else
      loop do
        job = attach(@backend.dequeue(topic))
        if job.nil?
          break if poll_interval <= 0
          sleep(poll_interval)
          next
        end
        yield job
        consumed += 1
        break if iterations > 0 && consumed >= iterations
      end
    end
  else
    Enumerator.new do |yielder|
      consumed = 0
      loop do
        job = attach(@backend.dequeue(topic))
        if job.nil?
          break if poll_interval <= 0
          sleep(poll_interval)
          next
        end
        yielder << job
        consumed += 1
        break if iterations > 0 && consumed >= iterations
      end
    end
  end
end

#dead_letters(max_retries: nil) ⇒ Object

Get dead letter jobs — messages that exceeded max retries. Pass max_retries to override the queue's default.



189
190
191
192
 
# File 'lib/tina4/queue.rb', line 189

def dead_letters(max_retries: nil) # -> list[dict]
  refuse_operation!("dead_letters()") unless @backend.respond_to?(:dead_letters)
  @backend.dead_letters(@topic, max_retries: max_retries || @max_retries)
end

#failed ⇒ Object

Get jobs that failed but are still eligible for retry (under max_retries).



175
176
177
178
 
# File 'lib/tina4/queue.rb', line 175

def failed # -> list[dict]
  refuse_operation!("failed()") unless @backend.respond_to?(:failed)
  @backend.failed(@topic, max_retries: @max_retries)
end

#get_topic ⇒ Object

Get the topic name this queue was constructed with.



358
359
360
 
# File 'lib/tina4/queue.rb', line 358

def get_topic
  @topic
end

#pop ⇒ Object

Pop the next available job. Returns Job or nil.



118
119
120
 
# File 'lib/tina4/queue.rb', line 118

def pop # -> Job|None
  attach(@backend.dequeue(@topic))
end

#pop_batch(count) ⇒ Object

Pop up to count jobs at once. Returns a partial batch if fewer available.



123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/tina4/queue.rb', line 123

def pop_batch(count)
  jobs =
    if @backend.respond_to?(:dequeue_batch)
      @backend.dequeue_batch(@topic, count)
    else
      collected = []
      count.times do
        job = @backend.dequeue(@topic)
        break if job.nil?
        collected << job
      end
      collected
    end
  jobs.each { |job| attach(job) }
  jobs
end

#pop_by_id(topic = nil, id) ⇒ Object

Pop a specific job by ID from the queue. Searches the pending queue for the given topic (defaults to this queue's topic). Returns the matching Job (claimed/removed from the queue) or nil.



320
321
322
323
324
325
326
327
328
329
 
# File 'lib/tina4/queue.rb', line 320

def pop_by_id(topic = nil, id)
  # Was `return nil` - indistinguishable from "no such job".
  unless @backend.respond_to?(:find_by_id)
    raise NotImplementedError,
          "The #{@backend.class.name.split('::').last} queue backend cannot " \
          "perform pop_by_id(): it cannot address a single message by id. Use " \
          "the file or mongodb backend."
  end
  attach(@backend.find_by_id(topic || @topic, id))
end

#process(topic: nil, max_jobs: nil, batch_size: 1, &handler) ⇒ Object

Consume all available jobs and pass each to handler, then stop.

Simpler alternative to consume() for drain-and-exit use cases.

queue.process { |job| handle(job); job.complete }
queue.process(topic: "emails", max_jobs: 10) { |job| ... }

Raises:

  • (ArgumentError) 


369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
 
# File 'lib/tina4/queue.rb', line 369

def process(topic: nil, max_jobs: nil, batch_size: 1, &handler)
  raise ArgumentError, "block required" unless block_given?
  drain_topic = topic || @topic
  processed = 0
  loop do
    break if max_jobs && processed >= max_jobs
    if batch_size > 1
      remaining = max_jobs ? [batch_size, max_jobs - processed].min : batch_size
      jobs = @backend.respond_to?(:dequeue_batch) ?
               @backend.dequeue_batch(drain_topic, remaining) :
               (1..remaining).map { @backend.dequeue(drain_topic) }.compact
      break if jobs.empty?
      begin
        handler.call(jobs)
      rescue => e
        jobs.each { |job| job.fail(e.message) }
      end
      processed += jobs.length
    else
      job = @backend.dequeue(drain_topic)
      break if job.nil?
      begin
        handler.call(job)
      rescue => e
        job.fail(e.message)
      end
      processed += 1
    end
  end
end

#produce(topic, payload, priority: 0, delay_seconds: 0) ⇒ Object

Produce a message onto a topic. Convenience wrapper around push().



226
227
228
229
230
231
 
# File 'lib/tina4/queue.rb', line 226

def produce(topic, payload, priority: 0, delay_seconds: 0)
  available_at = delay_seconds > 0 ? Time.now + delay_seconds : nil
  message = Job.new(topic: topic, payload: payload, priority: priority, available_at: available_at, queue: self)
  @backend.enqueue(message)
  message
end

#purge(status, max_retries: nil) ⇒ Object

Delete messages by status (completed, failed, dead).



195
196
197
198
 
# File 'lib/tina4/queue.rb', line 195

def purge(status, max_retries: nil) # -> int
  refuse_operation!("purge()") unless @backend.respond_to?(:purge)
  @backend.purge(@topic, status)
end

#push(payload, priority: 0, delay_seconds: 0) ⇒ Object

Push a job onto the queue. Returns the Job. priority: higher-priority messages are dequeued first (default 0). delay_seconds: delay before the message becomes available (default 0).



110
111
112
113
114
115
 
# File 'lib/tina4/queue.rb', line 110

def push(payload, priority: 0, delay_seconds: 0)
  available_at = delay_seconds > 0 ? Time.now + delay_seconds : nil
  message = Job.new(topic: @topic, payload: payload, priority: priority, available_at: available_at, queue: self)
  @backend.enqueue(message)
  message
end

#refuse_operation!(operation) ⇒ Object

A backend that does not implement an operation must SAY SO, naming itself and the operation (invariant 6).

These five methods used to return [] / false / 0 unless @backend.respond_to?(...). That turned "this backend cannot do this" into "nothing has failed" / "nothing needed retrying" / "nothing was purged" -- answers indistinguishable from a genuine empty success (ADR-0022 decision 7). It is the silent-no-op class this invariant exists to remove, and it hid every missing broker method behind a plausible-looking result.

Raises:

  • (NotImplementedError) 


216
217
218
219
220
221
222
223
 
# File 'lib/tina4/queue.rb', line 216

def refuse_operation!(operation)
  name = @backend.class.name.to_s.split("::").last.sub(/Backend\z/, "").downcase
  raise NotImplementedError,
        "The #{name} queue backend cannot perform #{operation}: it does not " \
        "implement that operation. Returning an empty result would be " \
        "indistinguishable from a successful empty answer. Use the file or " \
        "mongodb backend, which implement the full failure lifecycle."
end

#retry(job_id = nil, delay_seconds: 0) ⇒ Object

Retry a specific failed job by ID, or all dead-letter jobs if no id given. Returns true if re-queued.



182
183
184
185
 
# File 'lib/tina4/queue.rb', line 182

def retry(job_id = nil, delay_seconds: 0) # -> bool
  refuse_operation!("retry()") unless @backend.respond_to?(:retry_job)
  @backend.retry_job(@topic, job_id: job_id, delay_seconds: delay_seconds)
end

#retry_failed(max_retries: nil) ⇒ Object

Re-queue failed messages (under max_retries) back to pending. Returns the number of jobs re-queued.



202
203
204
205
 
# File 'lib/tina4/queue.rb', line 202

def retry_failed(max_retries: nil) # -> int
  refuse_operation!("retry_failed()") unless @backend.respond_to?(:retry_failed)
  @backend.retry_failed(@topic, max_retries: max_retries || @max_retries)
end

#size(status: "pending") ⇒ Object

Get the number of messages by status. status: "pending" (default) counts pending messages in the topic queue. status: "failed" or "dead" counts messages in the dead_letter directory.



334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
 
# File 'lib/tina4/queue.rb', line 334

def size(status: "pending")
  case status.to_s
  when "pending"
    @backend.size(@topic)
  when "reserved"
    @backend.respond_to?(:reserved_count) ? @backend.reserved_count(@topic) : 0
  when "failed", "dead"
    if @backend.respond_to?(:dead_letter_count)
      @backend.dead_letter_count(@topic)
    else
      0
    end
  when "completed"
    # Terminal-completed jobs. The lite/file backend deletes on complete
    # (no completed store) so this is 0; backends that track completion
    # expose #completed_count. Parity with the Python master's
    # size("completed") ("0 on the file backend") — never the pending count.
    @backend.respond_to?(:completed_count) ? @backend.completed_count(@topic) : 0
  else
    @backend.size(@topic)
  end
end