XDR transaction-queue-limit sizing

Audience: DBAs and SREs who run Aerospike XDR.
Applies to: Aerospike Database Enterprise 5.7 to 8.2. The mechanics described here are the same in all of these versions.

Symptom

On a source cluster, one or more of the following happens for a DC:

• recoveries jumps by about one per master partition when a destination comes back after an outage or slowdown, and the source nodes then read heavily from storage.
• in_queue climbs steadily and then drops sharply, while throughput stays the same or is 0. This repeats in a sawtooth.
• lag keeps rising, and xdr-dc-state shows master partitions in state B instead of T.

The usual cause is a partition's XDR transaction queue reaching transaction-queue-limit. At the default log level, the server writes no log line when this happens, so the counters and partition states above are the only evidence. Section 5.4 shows how to log discards.

1. What the transaction queue is

XDR keeps one transaction queue for each combination of DC × namespace × partition, on every node that holds a copy of the partition. Replica copies have queues too, not only masters.

• Each write that the DC ships adds one entry to the queue. This includes client writes, deletes, UDF writes, touches and transaction roll-forwards. On the master, the write itself adds the entry. On each replica, the replica write adds it.
• An entry holds no record data. It is the record's 20-byte digest plus a 5-byte last-update time: 25 bytes.
• A write is never queued for a DC that would not ship it: for example, a set excluded by ship-only-specified-sets or ignore-set, an XDR-written record with forward false, or an expunge with ignore-expunges true.
• If a key is written again within hot-key-ms (default 100 ms) while its earlier entry is still queued, the server adds no new entry. The hot_keys statistic counts these.

On the master, XDR drains the queue in laps. One lap runs every period-ms (default 100 ms). In each lap, a partition can send records until it has 50 in flight. A partition with no free slots sends nothing in that lap.

On a replica, nothing is shipped. The replica keeps its entries until the master reports that it has shipped past them. The master publishes this point, its safe LST, every 2 seconds, and the safe LST trails the shipped point by a 3-second safety margin. So a replica's queue holds 3 to 5 seconds more writes than its master's, about 4 seconds on average. This lets a replica that becomes master carry on shipping without a recovery.

2. What happens when a queue reaches the limit

Once per lap, XDR compares each partition's queue depth with transaction-queue-limit. If the depth is equal to or greater than the limit:

1. The whole queue is discarded, not just the excess. Its memory is freed.
2. On the master, the partition moves to pre-recovery (state B in xdr-dc-state).
   • The partition queues its recovery only after every shipment it has in flight has finished.
   • While the destination is unreachable, those shipments cannot finish. The partition stays in B, its new queue fills up again, and it is discarded again each time it reaches the limit.
   • recoveries_pending counts a partition only once its recovery is queued, when the partition leaves state B. So during an outage, recoveries_pending stays at 0 however many queues have been discarded.
3. Recovery scans the partition's primary index and ships every record whose last-update time is at or after the partition's safe LST. When it finishes, recoveries goes up by one. The cost of this:
   • no data is lost, but records are read from storage instead of being taken from the queue;
   • lag stays high for that partition until the recovery finishes;
   • while any recovery is pending for the DC and namespace, one quarter of the node's max-throughput quota is reserved for recovery. Transaction shipping for all partitions on the node then has only three quarters of the quota. That can push other partitions over the limit, so the problem spreads.
4. On a replica, the partition is marked incomplete (state I). This has no cost unless the replica becomes master before the master's next safe-LST update. In that case, the new master runs a recovery.

The limit cannot fix a sustained overload. If writes arrive at a partition faster than it can ship them, its queue keeps growing. A larger limit only delays the discard. A larger limit is the right tool for short stalls, such as a destination restart, a network blip or a connector pause. Section 4.3 covers sustained overload.

3. Memory: what a given limit costs

Size the worst case per node, per DC, per namespace:

memory = (4096 × replication-factor ÷ number-of-nodes) × 25 bytes × transaction-queue-limit

• 4096 × replication-factor ÷ number-of-nodes is the number of partition copies, master and replica, that one node holds. Each copy has its own queue.
• Add up the result for every DC and every namespace that ships to it. Each DC has its own queues.
• Add 12% for the memory allocator. A queue's size is always 25 bytes × a power of 2, and the allocator rounds every such size up by 12%. Budget 28 bytes per entry for the memory the process actually uses.

How to read these numbers:

• They are a ceiling, not a reservation. A queue starts at 64 entries and doubles in size as it fills. Memory therefore grows in steps: it jumps when queues pass a power of 2.
• A queue never shrinks when it drains. It keeps the largest size it reached until it is discarded: on overflow, when the limit is lowered below its depth, or when the node stops holding the partition. Memory therefore follows the peak depth since the last discard, not the current depth.
• The worst case is a realistic one. While a destination is down, every master and replica queue for that DC fills at the same rate. All of them reach the limit together.
• Allow up to 2× briefly. A queue that passes the limit between two lap checks doubles its size once before it is discarded.
• Allow for node loss. When a node leaves, the remaining nodes take over its partitions. Use number-of-nodes − 1, or the largest per-node partition count from show pmap, if memory headroom is tight.
• The memory is process heap. It is counted in the node's heap_allocated_kbytes statistic and is not part of any namespace's memory budget.

4. How large the limit needs to be

4.1 Arrival rate per partition

Each node's master partitions share that node's writes. So for one DC, on each node:

λ (writes/s per partition) = node's master write rate ÷ master partitions on that node

Master partitions on each node: the Primary column of:

asadm -e "show pmap"

Master write rate on each node: the XDR throughput statistic, in records per second averaged over 10 seconds. One column per node, plus a total column from -t, for every DC that ships the namespace:

asadm -e "show statistics xdr namespace for test like throughput -t"

• With an asadm version that has no show statistics xdr namespace command, run the get-stats command from section 5.1 on each node and read throughput.
• While XDR is keeping up (lag near 0, in_queue not growing), throughput equals the rate at which the node's master queues receive entries, after set filters and hot-key merging. That is exactly the rate λ needs.
• During an outage it reads 0, and during a recovery it reads higher than the write rate. In those cases, use the write row of asadm -e "show latencies" (ops/sec per node) as an estimate instead. It counts single-record client writes, before any XDR set filters, including writes that arrive from other DCs through XDR. Deletes, UDF writes and batch writes are not in it. So it can be lower than the shipped rate if your application uses those, and higher if the cluster also receives XDR writes that it does not forward.

Use the highest λ across nodes. If partitions and traffic are spread evenly, every node gives about the same answer as the cluster-wide shortcut:

λ = W ÷ 4096

where W is the total column above: the cluster-wide rate of writes this DC ships. Use master writes, not writes × RF: each replica's queue receives the same rate as its master's. Hot keys concentrate writes on a few partitions, and hot-key-ms merges only repeated writes of the same key, so budget for your busiest partition if your traffic has hot spots. Even with uniform traffic, the busiest partitions reach the limit somewhat before the average one does.

4.2 Time to overflow

When a partition stops draining (destination down, connector stalled, throttled retries), its queue fills at λ:

T_overflow (seconds) = transaction-queue-limit ÷ λ

A replica overflows up to about 5 seconds sooner than its master, because it already holds that much (section 1).

To ride out a stall of T_target seconds without recoveries:

transaction-queue-limit ≥ λ × (T_target + 5)

Round up to the next power of 2. The valid range is 1024 to 1048576 (1024 × 1024). Then check the memory cost with section 3.

Worked example: 20 nodes, RF 2, two DCs, 1,000,000 writes/s, and you want to survive a 5-minute destination outage.

• λ = 1,000,000 ÷ 4096 = 244/s
• 244 × (300 + 5) = 74,420 → round up to 131072, which allows about 9 minutes
• Memory: (4096 × 2 ÷ 20) × 28 × 131072 = 1.4 GiB per DC per namespace. With 2 DCs that is 2.8 GiB per node. Allow up to 2× briefly at an overflow, and about 5% more if one node is lost (4096 × 2 ÷ 19).

4.3 Drain ceilings: where a larger limit cannot help

A queue also grows, with no destination problem at all, if writes arrive faster than XDR can ship them. Two ceilings apply:

Per node, the per-partition ceiling adds up to about (4096 ÷ number-of-nodes) × 500 records/s. This falls as the cluster grows: about 341,000/s per node on 6 nodes, but about 41,000/s per node on 50 nodes.

If a partition's λ is close to either ceiling, its queue never drains and keeps overflowing whatever the limit. The fixes for that are different:

• raise max-throughput, if the destination can take more;
• reduce ship latency (network, destination load, connector);
• lower period-ms (laps run more often);
• reduce the rate of shipped writes, for example with set filtering.

5. Detecting that you are close to the limit

Collect these from every source node. All are per DC and per namespace. Replace DC1 and test with your DC and namespace names. The appendix has an example script that runs these checks for every node and prints a summary.

5.1 Node statistics

asinfo -v "get-stats:context=xdr;dc=DC1;namespace=test" | tr ';' '\n' | grep -E '^(lag|in_queue|in_progress|throughput|recoveries|recoveries_pending|hot_keys)='

The same values appear every 10 seconds in the log, in the {test} xdr-dc DC1: ticker line (lag, throughput, in-queue, recoveries (done,pending)).

Count the node's master partitions for the namespace with asadm -e "show pmap" (the Primary column). Then:

average master queue depth = in_queue ÷ master partitions
time headroom (seconds)    ≈ transaction-queue-limit ÷ λ − lag

5.2 Deepest single partition

An average hides one hot or stuck partition. xdr-dc-state lists every partition's queue depth (the trans_q field) for both master (M) and replica (P) copies. This prints the deepest queue per role as a percentage of the limit. Set limit to your configured value:

asinfo -v "xdr-dc-state:dc=DC1;namespace=test" \
  | grep -oE '[ 0-9]{4}:[A-Z]:[A-Z]:[0-9_.]+:[0-9_.]+: *[0-9]+' \
  | awk -F: -v limit=16384 '{ gsub(/ /, "", $1); gsub(/ /, "", $6); if ($6 + 0 >= max[$2]) { max[$2] = $6 + 0; pid[$2] = $1 } } END { for (r in max) printf "role %s: deepest trans_q %d (partition %s) = %.1f%% of limit\n", r, max[r], pid[r], 100 * max[r] / limit }'

A line for role X covers partitions this node does not hold. Ignore it.

Releases before 7.2.0.22, 8.0.0.20, 8.1.2.5 and 8.2.0.0 cut each xdr-dc-state row at 100 bytes. This includes every 5.x, 6.x, 7.0 and 7.1 release. What it breaks depends on the length of the namespace name:

• 6 characters or fewer: nothing.
• 7 to 21 characters: the ; between rows is lost. The commands in sections 5.2 and 5.3 do not rely on it, so they still give correct results.
• 22 characters or more: the end of each trans_q value is cut off. It reads lower than the real depth (for example, 98 reads as 9) or is missing, and there is no error. The command in this section then under-reports, so use the average from section 5.1 instead: in_queue is not affected. The state count in section 5.3 still works.

Read your configured limit with:

asinfo -v "get-config:context=xdr;dc=DC1;namespace=test" | tr ';' '\n' | grep transaction-queue-limit

5.3 Partition states: has a queue already been discarded?

This counts partitions by role and state:

asinfo -v "xdr-dc-state:dc=DC1;namespace=test" | grep -oE '[ 0-9]{4}:[MP]:[A-Z]:' | awk -F: '{ c[$2 ":" $3]++ } END { for (k in c) print k, c[k] }'

5.4 Logging each discard

At the default info level, nothing is logged when a queue is discarded. With the xdr log context at detail, each partition logs one line when its queue is discarded:

{test:3557} DC DC1 role M -> M, state TRANSACTION -> PRE_RECOVERY

• On a replica, the line has role P and ends state TRANSACTION -> RECOVERY_INCOMPLETE.
• If the partition changes role at the same moment, the line shows both roles, for example role P -> M.
• Only the first discard is logged. While the partition waits in state B, further discards of its queue are not logged again.
• Lines with the new role X are partitions leaving this node, not discards. The count command below leaves them out.

The setting is dynamic and per node, and it is lost on restart. Find the id of the log sink, set xdr to detail, and set it back to info when you are done:

asinfo -v "logs"
asinfo -v "log-set:id=0;xdr=detail"
asinfo -v "log-set:id=0;xdr=info"

Count the discards in the log (adjust the path to your log file):

grep -cE "role . -> [MP], state TRANSACTION -> (PRE_RECOVERY|RECOVERY_INCOMPLETE)" /var/log/aerospike/aerospike.log

detail adds little else to the xdr context unless trace-sample is set: a few DC and LST housekeeping lines, and one line per shipment result (complete result, requeue result, abandon result) that the log folds into one (repeated:N) line per interval.

5.5 Alert thresholds

Alert 1: a queue is more than half full

• Condition: the deepest trans_q (section 5.2) is above 50% of the limit, or lag is above 50% of T_overflow (section 4.2).
• Meaning: if the current stall lasts as long again, queues overflow and partitions go to recovery.
• Action: find the cause of the slow drain: the destination, the network, or a drain ceiling (section 4.3). If the stall is short and expected, raise the limit (section 6).

Alert 2: XDR cannot keep up with the write rate

• Condition: in_queue rises steadily while throughput stays flat, either at max-throughput or at about 500 × the node's master partitions.
• Meaning: shipping is drain-bound. The queues will overflow whatever the limit is.
• Action: remove the bottleneck (section 4.3). A larger limit only delays the overflow.

Alert 3: queues are being discarded

• Condition: any master partitions in state B (section 5.3), TRANSACTION -> PRE_RECOVERY lines in the log with detail logging on (section 5.4), or an in_queue sawtooth: a steady climb, then a sudden drop with no rise in throughput.
• Meaning: partition queues are reaching the limit and being discarded. While the destination is unreachable, recoveries_pending stays at 0, so this is the signal to watch.
• Action: fix the cause first, as for alerts 1 and 2. If the cause was a short stall, raise the limit so the next one fits.

Alert 4: recoveries without a cluster event

• Condition: recoveries increases, and no node has started, joined or left, no migration has run, and no rewind or DC/namespace add has been done. Those events cause recoveries normally.
• Meaning: discarded queues are being recovered: records are re-read from storage and lag stays high until the recoveries finish.
• Action: the same as alert 3.

6. Changing the limit

The parameter is set per DC per namespace, on each node. It can be changed at runtime:

asinfo -v "set-config:context=xdr;dc=DC1;namespace=test;transaction-queue-limit=131072"

The value must be a power of 2 from 1024 to 1048576. Any other value is rejected with ERROR:4, and the log shows invalid 'transaction-queue-limit'. Run the command on every source node. Also add the value to aerospike.conf so it survives a restart:

xdr {
    dc DC1 {
        # keep the existing node-address-port lines and other settings
        namespace test {
            transaction-queue-limit 131072
        }
    }
}

In aerospike.conf, a value that is not a power of 2 stops the node from starting, with transaction-queue-limit must be an exact power of 2 in the log.

• Raising the limit takes effect at the next lap. Memory grows only as queues actually get deeper.
• Lowering the limit discards, within a lap, every queue that is already at or above the new value. Each of those master partitions then needs a recovery. Lower it only when the queues are shallow (section 5.2), and never during a stall.

Appendix: example report script

python3 xdr-tq-report.py --dc DC1 --ns test 10.0.0.1 10.0.0.2 10.0.0.3

== 10.0.0.2  build 8.1.2.4-4  dc DC1  namespace test
  limit 1024  master partitions 2048  replica partitions 2048
  in_queue 1527836 -> 755133  lag 172 s  throughput 0/s  recoveries 2048 -> 2048  pending 0
  deepest trans_q: master 1024 (100.0% of limit), replica 1023 (99.9% of limit)
  states: M:B=1385 M:T=663 P:I=1614 P:T=434
  lambda 6.14/s per partition (write ops/sec (estimate))  T_overflow 167 s  headroom -5 s
  memory ceiling 112 MiB for this DC and namespace (4096 queues x 1024 x 28 B)
  ALERT 1: a queue is more than half full
  ALERT 3: queues are being discarded (1385 master partitions in state B)

#!/usr/bin/env python3
"""Example only: XDR transaction queue report. Not supported tooling; review and adapt before use.

Runs the asinfo commands from this article against each source node, for one DC and one namespace,
takes two samples INTERVAL seconds apart, and prints per node: queue depths, partition states,
arrival rate per partition, time to overflow, memory ceiling, and the alert checks.

usage: python3 xdr-tq-report.py --dc DC1 --ns test [--interval 10] [--asinfo-args "-U admin -P secret"] HOST[:PORT] ...
List every source node. Needs asinfo on the PATH and Python 3.6 or later.
"""
import argparse
import re
import shlex
import subprocess
import sys
import time

ROW = re.compile(r'[ 0-9]{4}:([A-Z]):([A-Z]):[0-9_.]+:[0-9_.]+: *([0-9]+)')  # role, state, trans_q
STATE = re.compile(r'[ 0-9]{4}:([MP]):([A-Z]):')                              # role, state
BYTES_PER_ENTRY = 28  # 25-byte entry plus the allocator's 12% rounding


def asinfo(node, cmd, extra):
    host, _, port = node.partition(':')
    args = ['asinfo', '-h', host, '-p', port or '3000'] + extra + ['-v', cmd]
    res = subprocess.run(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                         universal_newlines=True, timeout=120)
    out = res.stdout.strip()
    if res.returncode != 0 or out.startswith('ERROR'):
        raise RuntimeError('{}: {}: {}'.format(node, cmd, (out + res.stderr).strip()[:200]))
    return out


def kv(text):
    return dict(f.split('=', 1) for f in text.split(';') if '=' in f)


def sample(node, dc, ns, extra):
    s = kv(asinfo(node, 'get-stats:context=xdr;dc={};namespace={}'.format(dc, ns), extra))
    stats = {k: int(s.get(k, 0)) for k in
             ('lag', 'in_queue', 'throughput', 'recoveries', 'recoveries_pending')}
    return stats, asinfo(node, 'xdr-dc-state:dc={};namespace={}'.format(dc, ns), extra)


def write_ops(node, ns, extra):
    # Single-record client writes per second, including writes that arrive through XDR: an estimate only.
    out = asinfo(node, 'latencies:hist={%s}-write' % ns, extra)
    try:
        return float(out.split(':', 1)[1].split(',')[1])
    except (IndexError, ValueError):
        return 0.0


def report(node, a):
    extra = shlex.split(a.asinfo_args)
    build = asinfo(node, 'build', extra)
    ns_cfg = kv(asinfo(node, 'get-config:context=xdr;dc={};namespace={}'.format(a.dc, a.ns), extra))
    dc_cfg = kv(asinfo(node, 'get-config:context=xdr;dc={}'.format(a.dc), extra))
    limit = int(ns_cfg['transaction-queue-limit'])
    max_tput = int(ns_cfg.get('max-throughput', 100000))
    period_ms = int(dc_cfg.get('period-ms', 100))

    s1, _ = sample(node, a.dc, a.ns, extra)
    time.sleep(a.interval)
    s2, raw = sample(node, a.dc, a.ns, extra)

    states, deepest, m_sum, rows = {}, {}, 0, 0
    for role, state in STATE.findall(raw):
        states[role + ':' + state] = states.get(role + ':' + state, 0) + 1
    for role, state, depth in ROW.findall(raw):
        depth = int(depth)
        rows += 1
        if role == 'M':
            m_sum += depth
        if role in 'MP' and depth >= deepest.get(role, -1):
            deepest[role] = depth
    masters = sum(v for k, v in states.items() if k.startswith('M:'))
    replicas = sum(v for k, v in states.items() if k.startswith('P:'))
    n_b = states.get('M:B', 0)

    # Releases before 7.2.0.22, 8.0.0.20, 8.1.2.5 and 8.2.0.0 cut trans_q for namespace names of 22+ characters.
    depth_ok = rows >= masters + replicas and not (
        s2['in_queue'] > 1000 and m_sum < s2['in_queue'] // 2)

    # throughput equals the arrival rate only while XDR keeps up: it is 0 during an outage and
    # inflated while recoveries catch up.
    healthy = (s1['lag'] <= 5 and s2['lag'] <= 5 and s2['recoveries_pending'] == 0 and n_b == 0 and
               s2['recoveries'] == s1['recoveries'] and s2['throughput'] > 0)
    if masters and healthy:
        lam, lam_src = s2['throughput'] / masters, 'xdr throughput'
    elif masters:
        lam, lam_src = write_ops(node, a.ns, extra) / masters, 'write ops/sec (estimate)'
    else:
        lam, lam_src = 0.0, 'no master partitions'
    t_ovf = limit / lam if lam > 0 else None

    per_part_ceiling = 50 * 1000 / period_ms
    alerts = []
    worst = max(deepest.values()) if deepest and depth_ok else None
    if (worst is not None and worst > limit / 2) or (t_ovf and s2['lag'] > t_ovf / 2):
        alerts.append('ALERT 1: a queue is more than half full')
    if s2['in_queue'] > s1['in_queue'] and s2['throughput'] > 0 and (
            s2['throughput'] >= 0.95 * max_tput or
            s2['throughput'] >= 0.95 * per_part_ceiling * masters):
        alerts.append('ALERT 2: shipping is drain-bound (in_queue rising at a throughput ceiling)')
    if n_b or (s1['in_queue'] > 1000 and s2['in_queue'] < s1['in_queue'] // 2 and
               s2['throughput'] <= s1['throughput']):
        alerts.append('ALERT 3: queues are being discarded ({} master partitions in state B)'.format(n_b))
    if s2['recoveries'] > s1['recoveries']:
        alerts.append('ALERT 4: recoveries +{} in {} s (rule out node start/join/leave, migration, rewind, DC add)'
                      .format(s2['recoveries'] - s1['recoveries'], a.interval))
    if not depth_ok:
        alerts.append('NOTE: xdr-dc-state trans_q is unreliable on this build (section 5.2); use in_queue')

    def pct(v):
        return '{} ({:.1f}% of limit)'.format(v, 100.0 * v / limit) if v is not None and depth_ok else 'n/a'

    print('== {}  build {}  dc {}  namespace {}'.format(node, build, a.dc, a.ns))
    print('  limit {}  master partitions {}  replica partitions {}'.format(limit, masters, replicas))
    print('  in_queue {} -> {}  lag {} s  throughput {}/s  recoveries {} -> {}  pending {}'.format(
        s1['in_queue'], s2['in_queue'], s2['lag'], s2['throughput'],
        s1['recoveries'], s2['recoveries'], s2['recoveries_pending']))
    print('  deepest trans_q: master {}, replica {}'.format(pct(deepest.get('M')), pct(deepest.get('P'))))
    print('  states: ' + ' '.join('{}={}'.format(k, v) for k, v in sorted(states.items())))
    if t_ovf:
        print('  lambda {:.2f}/s per partition ({})  T_overflow {:.0f} s  headroom {:.0f} s'.format(
            lam, lam_src, t_ovf, t_ovf - s2['lag']))
    else:
        print('  lambda unknown ({})'.format(lam_src))
    print('  memory ceiling {:.0f} MiB for this DC and namespace ({} queues x {} x {} B)'.format(
        (masters + replicas) * limit * BYTES_PER_ENTRY / 2.0 ** 20, masters + replicas, limit, BYTES_PER_ENTRY))
    for line in alerts or ['OK: no alert conditions']:
        print('  ' + line)
    return lam, t_ovf, s2['lag']


def main():
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument('--dc', required=True)
    p.add_argument('--ns', required=True)
    p.add_argument('--interval', type=int, default=10, help='seconds between samples (default 10)')
    p.add_argument('--asinfo-args', default='', help='extra asinfo options, e.g. "-U user -P pass" or TLS options')
    p.add_argument('nodes', nargs='+', help='every source node, as host or host:port')
    a = p.parse_args()

    results = []
    for node in a.nodes:
        try:
            results.append(report(node, a))
        except Exception as e:  # keep going for the other nodes
            print('== {}  FAILED: {}'.format(node, e))
    known = [r for r in results if r[1]]
    if known:
        lam, t_ovf, lag = max(known, key=lambda r: r[0])
        print('\nHighest lambda {:.2f}/s per partition: T_overflow {:.0f} s, headroom {:.0f} s'.format(
            lam, t_ovf, t_ovf - lag))
        print('To ride out a stall of T seconds: transaction-queue-limit >= {:.2f} x (T + 5), '
              'rounded up to a power of 2'.format(lam))


if __name__ == '__main__':
    sys.exit(main())

References

• XDR configuration reference: transaction-queue-limit
• XDR configuration reference: max-throughput
• XDR configuration reference: period-ms
• XDR metrics reference: in_queue, lag, recoveries_pending
• XDR record shipment lifecycle