#!/usr/bin/env python3
"""SIGKILL after sender SM acknowledgement under real delayed durable IO.
Four STARTTLS resources/two accounts, MAM, carbons and unacknowledged avatar
PEP. Inspect authoritative recipient journals and both archives before any
reauthentication; then resume all resources and verify replay and counters.
Uses the latency fixture's fsync/fdatasync interposer, retaining real syncs.
"""
import argparse
import asyncio
import hashlib
import json
import os
from pathlib import Path
import shutil
import signal
import socket
import ssl
import subprocess
import sys
import tempfile
import time
from muc_sm_load_network import SM
from muc_restart_network import stop
from sm_restart_network import durable_slot
from slow_storage_latency_network import (
CARBONS, PUBSUB, MeasuredClient, check_shim_scope, counts, iq, sync_rows,
)
def start(binary, config, port, log, env):
server = subprocess.Popen([str(binary), 'serve', str(config), '--experimental'],
env=env, stdout=log, stderr=log)
deadline = time.monotonic()+60
while time.monotonic() < deadline:
assert server.poll() is None, 'server stopped during startup'
try:
with socket.create_connection(('127.0.0.1', port), timeout=.1):
return server
except OSError:
time.sleep(.05)
server.kill()
server.wait()
raise AssertionError('server startup timeout')
def committed_archive(path):
"""Reconstruct native MAM bytes using its authoritative checksummed redo."""
native = path.read_bytes()
data = path.with_suffix('.rredo').read_bytes()
assert len(data) >= 96 and data[:8] == b'XARRED01', 'invalid MAM redo header'
assert hashlib.sha256(data[:64]).digest() == data[64:96], 'invalid MAM redo header checksum'
baseline = int.from_bytes(data[8:16], 'little')
assert baseline <= len(native), 'missing MAM native baseline'
committed, offset = native[:baseline], 96
while offset < len(data):
if len(data)-offset < 72:
break
size = int.from_bytes(data[offset:offset+8], 'little')
assert hashlib.sha256(data[offset:offset+8]).digest() == data[offset+8:offset+40], 'invalid MAM redo length checksum'
assert 0 < size <= 16*1024*1024, 'invalid MAM redo frame size'
end = offset+72+size
if end > len(data):
break
payload = data[offset+72:end]
assert hashlib.sha256(payload).digest() == data[offset+40:offset+72], 'invalid MAM redo frame checksum'
committed += payload
offset = end
# Sender handled acknowledgement must not depend on a native suffix that
# lacks a complete, durable redo frame.
assert native == committed, 'native MAM suffix is not committed in redo'
return committed.decode()
def assert_committed(root, expected, identifiers, avatar):
snapshots = [durable_slot(root, old['token']) for old in expected]
for index, (record, old) in enumerate(zip(snapshots, expected)):
assert record['jid'] == old['jid'], (index, record)
assert record['handled'] == old['handled'], (index, record['handled'], old)
assert record['acked'] == old['acked'], (index, record['acked'], old)
assert record['sent']-record['acked'] == len(record['pending']), (index, record)
if index in (1, 2, 3):
for identifier in identifiers:
assert sum(identifier in xml for xml in record['pending']) == 1, (index, identifier, record)
if index in (0, 2):
assert any(avatar in xml and 'event' in xml and 'crash-avatar' in xml
for xml in record['pending']), ('PEP missing before reauthentication', index, record)
paths = sorted((root/'data'/'localhost'/'archive').glob('*.list'))
assert len(paths) == 2, paths
for path in paths:
rows = [row for row in committed_archive(path).splitlines() if row.startswith('item(')]
assert len(rows) == len(identifiers), (path.name, len(rows), identifiers)
for identifier in identifiers:
assert sum(identifier in row for row in rows) == 1, ('acknowledged body missing/duplicated in MAM', path.name, identifier)
return snapshots
async def exercise(root, binary, config, port, log, env, sync_log, args):
context = ssl.create_default_context(cafile=str(root/'cert.pem'))
clients, server = [], None
avatar = 'urn:xmpp:avatar:metadata'
try:
server = await asyncio.to_thread(start, binary, config, port, log, env)
for number in range(4):
client = await MeasuredClient.connect(number, port, context, 2)
clients.append(client)
client.auto_ack = False
await client.enable()
await iq(client, f'carbons-{number}', f"")
await client.stanza('')
await client.fence(f'ready-{number}')
alice, bob, alice_other, _ = clients
await iq(alice, 'avatar-initial', f"