feat: per-sink delivery, journal forwarding, and operational API

Persistence:
- one delivery state per sink, with claim, crash recovery, and index replacement
- retention for delivered and dead records
- LMDB writes moved off the event loop

Pipeline:
- one attempt per dispatch, with the next try scheduled by next_retry_at
- backpressure through max_queue_size
- reload re-reads YAML; SIGTERM drains and SIGHUP reloads

Plugins:
- journald source, enricher and filter processors, Loki sink
- built-ins register when entry points are missing

API and CLI:
- health, metrics, per-sink queue, and dead-letter replay
- startup prints the pipeline on one line

Tests:
- dispatch, plugins, metrics, CLI, and persistence coverage

tests/conftest.py

@@ -34,6 +34,7 @@ def sample_record() -> StoredRecord:
         record_type="test",
         body={"message": "hello"},
         metadata={"env": "test"},
+        sinks=["out"],
     )
 
 

tests/test_cli.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from typer.testing import CliRunner
+
+from daena.cli.main import app
+
+runner = CliRunner()
+
+
+def test_version_command() -> None:
+    result = runner.invoke(app, ["version"])
+    assert result.exit_code == 0
+    assert "daena v" in result.output
+
+
+def test_status_reports_connection_failure() -> None:
+    result = runner.invoke(app, ["--target", "127.0.0.1:1", "status"])
+    assert result.exit_code == 1
+    assert "Cannot connect" in result.output
+
+
+def test_records_rejects_unknown_kind() -> None:
+    result = runner.invoke(app, ["records", "other"])
+    assert result.exit_code == 2
+    assert "pending or dead" in result.output

tests/test_dispatch.py

@@ -0,0 +1,182 @@
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+import pytest
+import yaml
+
+from daena.config import (
+    DaenaConfig,
+    HealthConfig,
+    PipelineConfig,
+    ProcessorConfig,
+    SinkConfig,
+    SinkRetryConfig,
+    SourceConfig,
+    load_config,
+)
+from daena.core.runtime import DaenaRuntime
+from daena.persistence.models import RecordState
+
+
+def _runtime(tmp_path: Path, **overrides: object) -> DaenaRuntime:
+    config = DaenaConfig()
+    config.persistence.lmdb.path = str(tmp_path / "lmdb")
+    config.pipeline = PipelineConfig(batch_size=10, flush_interval=0.05, max_queue_size=10_000)
+    config.retention.interval_seconds = 3600
+    for key, value in overrides.items():
+        setattr(config, key, value)
+    return DaenaRuntime(config)
+
+
+@pytest.mark.asyncio
+async def test_retry_does_not_block_until_success(tmp_path: Path) -> None:
+    runtime = _runtime(tmp_path)
+    runtime.config.sources = [
+        SourceConfig(
+            name="once",
+            type="dummy",
+            options={"interval": 0.05, "count": 1, "message": "retry"},
+        )
+    ]
+    runtime.config.sinks = [
+        SinkConfig(
+            name="out",
+            type="dummy",
+            retry=SinkRetryConfig(max_attempts=3, min_delay=0, max_delay=0, multiplier=1),
+            options={"fail_times": 1},
+        )
+    ]
+    await runtime.start()
+    delivered = 0
+    for _ in range(40):
+        assert runtime.backend is not None
+        delivered = runtime.backend.count(RecordState.delivered)
+        if delivered == 1:
+            break
+        await asyncio.sleep(0.05)
+    dead = runtime.backend.count(RecordState.dead) if runtime.backend else -1
+    await runtime.stop()
+    assert delivered == 1
+    assert dead == 0
+
+
+@pytest.mark.asyncio
+async def test_sinks_keep_independent_delivery_state(tmp_path: Path) -> None:
+    runtime = _runtime(tmp_path)
+    runtime.config.sources = [
+        SourceConfig(name="once", type="dummy", options={"interval": 0.05, "count": 1})
+    ]
+    runtime.config.sinks = [
+        SinkConfig(name="good", type="dummy"),
+        SinkConfig(name="bad", type="dummy", options={"fail_times": 5, "permanent": True}),
+    ]
+    await runtime.start()
+    stored = None
+    for _ in range(40):
+        assert runtime.backend is not None
+        records = runtime.backend.list_by_state(RecordState.dead, limit=10)
+        if records and records[0].deliveries["good"].state == RecordState.delivered:
+            stored = records[0]
+            break
+        await asyncio.sleep(0.05)
+    await runtime.stop()
+    assert stored is not None
+    assert stored.deliveries["good"].state == RecordState.delivered
+    assert stored.deliveries["bad"].state == RecordState.dead
+
+
+@pytest.mark.asyncio
+async def test_backpressure_stops_ingest(tmp_path: Path) -> None:
+    runtime = _runtime(tmp_path)
+    runtime.config.pipeline = PipelineConfig(batch_size=10, flush_interval=0.05, max_queue_size=2)
+    runtime.config.sources = [
+        SourceConfig(name="fast", type="dummy", options={"interval": 0.01, "message": "bp"})
+    ]
+    runtime.config.sinks = [SinkConfig(name="slow", type="dummy", options={"delay": 30})]
+    await runtime.start()
+    await asyncio.sleep(0.4)
+    assert runtime.backend is not None
+    stored = runtime.backend.count()
+    await runtime.stop()
+    assert stored <= 2
+
+
+@pytest.mark.asyncio
+async def test_disabled_source_emits_nothing(tmp_path: Path) -> None:
+    runtime = _runtime(tmp_path)
+    runtime.config.sources = [
+        SourceConfig(name="off", type="dummy", enabled=False, options={"interval": 0.05})
+    ]
+    runtime.config.sinks = [SinkConfig(name="out", type="dummy")]
+    await runtime.start()
+    await asyncio.sleep(0.3)
+    assert runtime.backend is not None
+    total = runtime.backend.count()
+    await runtime.stop()
+    assert total == 0
+
+
+@pytest.mark.asyncio
+async def test_reload_rereads_file(tmp_path: Path) -> None:
+    path = tmp_path / "daena.yaml"
+    data = {
+        "persistence": {"lmdb": {"path": str(tmp_path / "lmdb")}},
+        "pipeline": {"batch_size": 10, "flush_interval": 0.05, "max_queue_size": 100},
+        "retention": {"interval_seconds": 3600},
+        "sources": [{"name": "live", "type": "dummy", "enabled": True, "options": {"interval": 1}}],
+        "sinks": [{"name": "out", "type": "dummy", "enabled": True}],
+    }
+    path.write_text(yaml.dump(data), encoding="utf-8")
+    runtime = DaenaRuntime(load_config(path), config_path=path)
+    await runtime.start()
+    data["sources"][0]["enabled"] = False
+    path.write_text(yaml.dump(data), encoding="utf-8")
+    await runtime.reload()
+    assert runtime.config.sources[0].enabled is False
+    assert runtime.running is True
+    await runtime.stop()
+
+
+@pytest.mark.asyncio
+async def test_health_degrades_when_delivery_stalls(tmp_path: Path) -> None:
+    runtime = _runtime(tmp_path)
+    runtime.config.health = HealthConfig(max_pending_age_seconds=0)
+    runtime.config.sources = [
+        SourceConfig(name="fast", type="dummy", options={"interval": 0.05, "count": 1})
+    ]
+    runtime.config.sinks = [SinkConfig(name="slow", type="dummy", options={"delay": 30})]
+    await runtime.start()
+    await asyncio.sleep(1.2)
+    report = runtime.health()
+    await runtime.stop()
+    assert report.status == "degraded"
+    assert report.healthy is False
+    assert report.oldest_pending_age_seconds >= 1
+
+
+@pytest.mark.asyncio
+async def test_filter_drops_record_before_persist(tmp_path: Path) -> None:
+    runtime = _runtime(tmp_path)
+    runtime.config.sources = [
+        SourceConfig(
+            name="once",
+            type="dummy",
+            options={"interval": 0.05, "count": 1, "message": "drop-me"},
+        )
+    ]
+    runtime.config.processors = [
+        ProcessorConfig(
+            name="drop",
+            type="filter",
+            options={"drop": {"body.message": "drop-me #1"}},
+        )
+    ]
+    runtime.config.sinks = [SinkConfig(name="out", type="dummy")]
+    await runtime.start()
+    await asyncio.sleep(0.3)
+    assert runtime.backend is not None
+    total = runtime.backend.count()
+    await runtime.stop()
+    assert total == 0

tests/test_grpc.py

@@ -8,18 +8,20 @@ import pytest
 from daena.api.grpc_server import DaenaGRPCServer
 from daena.api.pb2 import daena_pb2 as pb2
 from daena.api.pb2 import daena_pb2_grpc as pb2_grpc
-from daena.config import DaenaConfig, SinkConfig, SourceConfig
+from daena.config import DaenaConfig, PipelineConfig, SinkConfig, SourceConfig
 from daena.core.runtime import DaenaRuntime
+from daena.persistence.models import RecordState
 
 
 @pytest.fixture
-async def grpc_env():
+async def grpc_env(tmp_path):
     config = DaenaConfig()
     config.sources = [
         SourceConfig(name="dummy", type="dummy", options={"interval": 0.2, "message": "test"})
     ]
     config.sinks = [SinkConfig(name="log", type="dummy")]
-    config.persistence.lmdb.path = "/tmp/daena_grpc_test"
+    config.persistence.lmdb.path = str(tmp_path / "lmdb")
+    config.pipeline = PipelineConfig(batch_size=100, flush_interval=5.0, max_queue_size=10_000)
     config.api.grpc.port = 18642
 
     runtime = DaenaRuntime(config)
@@ -76,7 +78,7 @@ async def test_get_queue_state(grpc_env) -> None:
     await asyncio.sleep(1)
     resp = await stub.GetQueueState(pb2.Empty(), timeout=5)
     assert resp.total > 0
-    assert resp.pending > 0
+    assert resp.pending + resp.delivered > 0
 
 
 @pytest.mark.asyncio
@@ -98,6 +100,30 @@ async def test_list_sinks(grpc_env) -> None:
     assert resp.sinks[0].name == "log"
 
 
+@pytest.mark.asyncio
+async def test_replay_dead(grpc_env) -> None:
+    runtime, stub, _ = grpc_env
+    await asyncio.sleep(1)
+    assert runtime.backend is not None
+    stored = None
+    for state in (RecordState.pending, RecordState.delivered, RecordState.failed):
+        found = runtime.backend.list_by_state(state, limit=1)
+        if found:
+            stored = found[0]
+            break
+    assert stored is not None
+    for sink in stored.deliveries:
+        runtime.backend.dead(stored.record_id, "grpc replay", sink=sink)
+    resp = await stub.ReplayDead(
+        pb2.ReplayRequest(record_id=stored.record_id, limit=10),
+        timeout=5,
+    )
+    assert resp.replayed == 1
+    again = runtime.backend.get(stored.record_id)
+    assert again is not None
+    assert again.deliveries["log"].state == RecordState.pending
+
+
 @pytest.mark.asyncio
 async def test_reload_config(grpc_env) -> None:
     _, stub, _ = grpc_env

tests/test_http_sink.py

@@ -10,7 +10,7 @@ from typing import Any
 
 import pytest
 
-from daena.config import DaenaConfig, SinkConfig, SourceConfig
+from daena.config import DaenaConfig, PipelineConfig, SinkConfig, SourceConfig
 from daena.core.runtime import DaenaRuntime
 from daena.logutil import setup_logging
 from daena.persistence.models import RecordState
@@ -20,6 +20,20 @@ setup_logging(level="error", fmt="json")
 pytestmark = pytest.mark.asyncio
 
 
+def _fast_pipeline(config: DaenaConfig) -> None:
+    config.pipeline = PipelineConfig(batch_size=100, flush_interval=0.05, max_queue_size=10_000)
+
+
+async def _wait_until(predicate: Any, timeout: float = 3.0) -> bool:
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while loop.time() < deadline:
+        if predicate():
+            return True
+        await asyncio.sleep(0.05)
+    return False
+
+
 class _TestHTTPServer:
     """Minimal async HTTP server to receive forwarded records."""
 
@@ -74,11 +88,12 @@ async def test_http_sink_forwards_records() -> None:
         config.data_dir = str(tmpdir)
         config.log_dir = str(tmpdir)
         config.persistence.lmdb.path = str(data_dir / "lmdb")
+        _fast_pipeline(config)
         config.sources = [
             SourceConfig(
                 name="test_source",
                 type="dummy",
-                options={"interval": 0.2, "message": "http_sink_test"},
+                options={"interval": 0.05, "count": 1, "message": "http_sink_test"},
             )
         ]
         config.sinks = [
@@ -91,7 +106,7 @@ async def test_http_sink_forwards_records() -> None:
 
         runtime = DaenaRuntime(config)
         await runtime.start()
-        await asyncio.sleep(3.0)
+        await _wait_until(lambda: len(server.received) > 0)
         await runtime.stop()
 
     await server.stop()
@@ -119,11 +134,12 @@ async def test_http_sink_permanent_failure_on_404() -> None:
         config.data_dir = str(tmpdir)
         config.log_dir = str(tmpdir)
         config.persistence.lmdb.path = str(data_dir / "lmdb")
+        _fast_pipeline(config)
         config.sources = [
             SourceConfig(
                 name="test_source",
                 type="dummy",
-                options={"interval": 0.2, "message": "http_404_test"},
+                options={"interval": 0.05, "count": 1, "message": "http_404_test"},
             )
         ]
         config.sinks = [
@@ -140,11 +156,13 @@ async def test_http_sink_permanent_failure_on_404() -> None:
 
         runtime = DaenaRuntime(config)
         await runtime.start()
-        await asyncio.sleep(3.0)
+        assert runtime.backend is not None
+        backend = runtime.backend
+        await _wait_until(lambda: backend.count(RecordState.dead) > 0)
 
-        dead = runtime.backend.count(RecordState.dead)
-        failed = runtime.backend.count(RecordState.failed)
-        total = runtime.backend.count()
+        dead = backend.count(RecordState.dead)
+        failed = backend.count(RecordState.failed)
+        total = backend.count()
 
         await runtime.stop()
 
@@ -168,11 +186,12 @@ async def test_http_sink_delivery_result() -> None:
         config.data_dir = str(tmpdir)
         config.log_dir = str(tmpdir)
         config.persistence.lmdb.path = str(data_dir / "lmdb")
+        _fast_pipeline(config)
         config.sources = [
             SourceConfig(
                 name="test_source",
                 type="dummy",
-                options={"interval": 0.2, "message": "http_delivery_test"},
+                options={"interval": 0.05, "count": 1, "message": "http_delivery_test"},
             )
         ]
         config.sinks = [
@@ -185,9 +204,11 @@ async def test_http_sink_delivery_result() -> None:
 
         runtime = DaenaRuntime(config)
         await runtime.start()
-        await asyncio.sleep(3.0)
+        assert runtime.backend is not None
+        backend = runtime.backend
+        await _wait_until(lambda: backend.count(RecordState.delivered) > 0)
 
-        delivered = runtime.backend.count(RecordState.delivered)
+        delivered = backend.count(RecordState.delivered)
         await runtime.stop()
 
     await server.stop()

tests/test_layout.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from daena.config import (
+    DaenaConfig,
+    MetricsConfig,
+    ProcessorConfig,
+    SinkConfig,
+    SourceConfig,
+)
+from daena.core.layout import preview_message, render_layout
+
+
+def _config() -> DaenaConfig:
+    config = DaenaConfig()
+    config.persistence.lmdb.path = "/tmp/daena/data"
+    config.sources = [
+        SourceConfig(name="dummy", type="dummy", options={"interval": 2.0}),
+        SourceConfig(
+            name="system",
+            type="journald",
+            enabled=False,
+            options={"units": ["ssh.service"]},
+        ),
+    ]
+    config.processors = [
+        ProcessorConfig(name="host", type="enricher", options={"static": {"env": "local"}}),
+        ProcessorConfig(name="keep", type="filter", options={"drop": {"body.priority": "7"}}),
+    ]
+    config.sinks = [
+        SinkConfig(name="log", type="dummy"),
+        SinkConfig(
+            name="loki",
+            type="loki",
+            options={"url": "http://127.0.0.1:3100/loki/api/v1/push"},
+        ),
+    ]
+    config.logging.verbose = True
+    return config
+
+
+def test_layout_is_a_single_pipeline_line() -> None:
+    text = render_layout(_config(), "./daena.yaml")
+    assert "┌" not in text
+    assert "▼" not in text
+    flow = text.splitlines()[1]
+    assert flow.index("dummy") < flow.index("host")
+    assert flow.index("host") < flow.index("lmdb")
+    assert flow.index("lmdb") < flow.index("log")
+    assert "daena 0.1.0  ./daena.yaml" in text
+    assert "/tmp/daena/data" in text
+    assert "127.0.0.1:8642   127.0.0.1:8643" in text
+
+
+def test_layout_names_each_plugin_once() -> None:
+    text = render_layout(_config(), None)
+    flow = text.splitlines()[1]
+    assert flow == " → ".join(
+        [
+            "dummy (2s), system (journald, off)",
+            "host (enricher), keep (filter)",
+            "lmdb",
+            "log (dummy), loki (127.0.0.1:3100)",
+        ]
+    )
+    assert "daena 0.1.0  defaults" in text
+
+
+def test_layout_skips_empty_stages_and_disabled_metrics() -> None:
+    config = DaenaConfig()
+    config.api.metrics = MetricsConfig(enabled=False)
+    text = render_layout(config, None)
+    assert "→ lmdb" not in text
+    assert text.splitlines()[1] == "lmdb"
+    assert text.splitlines()[-1] == "127.0.0.1:8642"
+
+
+def test_preview_message_uses_message_then_clips() -> None:
+    assert preview_message({"message": "daena heartbeat #1"}) == "daena heartbeat #1"
+    assert preview_message({"message": "a\nb"}) == "a b"
+    assert preview_message({"counter": 1}) == '{"counter":1}'
+    clipped = preview_message({"message": "x" * 130}, limit=10)
+    assert clipped.endswith("…")
+    assert len(clipped) == 10

tests/test_metrics.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from daena.api.metrics import MetricsServer, MetricsSnapshot, render_prometheus
+
+
+def test_render_prometheus_includes_sink_state() -> None:
+    text = render_prometheus(
+        MetricsSnapshot(
+            running=True,
+            stored_records=3,
+            oldest_inflight_seconds=12,
+            deliveries={("central", "pending"): 2, ("central", "delivered"): 1},
+            sink_results={"central": {"delivered": 4, "failed": 1, "dead": 0}},
+        )
+    )
+    assert "daena_up 1" in text
+    assert "daena_records 3" in text
+    assert "daena_oldest_inflight_seconds 12" in text
+    assert 'daena_delivery_records{sink="central",state="pending"} 2' in text
+    assert 'daena_sink_deliveries_total{sink="central",result="failed"} 1' in text
+
+
+@pytest.mark.asyncio
+async def test_metrics_server_serves_text() -> None:
+    snapshot = MetricsSnapshot(
+        running=True,
+        stored_records=0,
+        oldest_inflight_seconds=0,
+        deliveries={},
+        sink_results={},
+    )
+    server = MetricsServer("127.0.0.1", 0, lambda: snapshot, lambda: {"status": "ok"})
+    await server.start()
+    assert server._server is not None
+    sockets = server._server.sockets
+    assert sockets is not None
+    port = sockets[0].getsockname()[1]
+    reader, writer = await asyncio.open_connection("127.0.0.1", port)
+    writer.write(b"GET /metrics HTTP/1.1\r\nHost: localhost\r\n\r\n")
+    await writer.drain()
+    data = await reader.read()
+    writer.close()
+    await writer.wait_closed()
+    await server.stop()
+    assert b"200 OK" in data
+    assert b"daena_up 1" in data

tests/test_persistence.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from pathlib import Path
+
 from daena.persistence.lmdb import LMDBBackend
 from daena.persistence.models import RecordState, StoredRecord
 
@@ -105,3 +107,109 @@ def test_ready_for_retry(lmdb_backend: LMDBBackend, sample_record: StoredRecord)
     lmdb_backend.nack(sample_record.record_id, "retry later", retry_delay=0)
     ready = lmdb_backend.ready_for_retry(limit=10)
     assert len(ready) == 1
+
+
+def test_index_keys_do_not_accumulate(
+    lmdb_backend: LMDBBackend, sample_record: StoredRecord
+) -> None:
+    lmdb_backend.put(sample_record)
+    assert lmdb_backend.index_count() == 1
+    lmdb_backend.nack(sample_record.record_id, "later", retry_delay=0)
+    assert lmdb_backend.index_count() == 2
+    lmdb_backend.ack(sample_record.record_id)
+    assert lmdb_backend.ready_for_retry() == []
+    assert lmdb_backend.pending() == []
+    assert lmdb_backend.count(RecordState.failed) == 0
+    assert lmdb_backend.count(RecordState.delivered) == 1
+    assert lmdb_backend.index_count() == 1
+
+
+def test_claim_and_recover(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    claimed = lmdb_backend.claim("out", limit=10)
+    assert len(claimed) == 1
+    assert claimed[0].deliveries["out"].state == RecordState.delivering
+    assert lmdb_backend.pending() == []
+    assert lmdb_backend.recover_inflight() == 1
+    pending = lmdb_backend.pending()
+    assert len(pending) == 1
+    assert pending[0].deliveries["out"].state == RecordState.pending
+
+
+def test_per_sink_ack_leaves_other_sink_pending(lmdb_backend: LMDBBackend) -> None:
+    record = StoredRecord(source="s", record_type="t", body={"n": 1}, sinks=["a", "b"])
+    lmdb_backend.put(record)
+    lmdb_backend.ack(record.record_id, sink="a")
+    stored = lmdb_backend.get(record.record_id)
+    assert stored is not None
+    assert stored.deliveries["a"].state == RecordState.delivered
+    assert stored.deliveries["b"].state == RecordState.pending
+    assert lmdb_backend.count(RecordState.pending) == 1
+    assert lmdb_backend.count(RecordState.delivered) == 1
+    assert lmdb_backend.pending(sink="a") == []
+    assert len(lmdb_backend.pending(sink="b")) == 1
+    removed = lmdb_backend.delete_old("2100-01-01T00:00:00+00:00", state=RecordState.delivered)
+    assert removed == 0
+    assert lmdb_backend.get(record.record_id) is not None
+
+
+def test_replay_dead_becomes_pending(
+    lmdb_backend: LMDBBackend, sample_record: StoredRecord
+) -> None:
+    lmdb_backend.put(sample_record)
+    lmdb_backend.dead(sample_record.record_id, error="nope")
+    replayed = lmdb_backend.replay(record_id=sample_record.record_id)
+    stored = lmdb_backend.get(sample_record.record_id)
+    assert replayed == 1
+    assert stored is not None
+    assert stored.deliveries["out"].state == RecordState.pending
+    assert stored.deliveries["out"].retry_count == 0
+    assert stored.deliveries["out"].last_error is None
+
+
+def test_delete_expired_keeps_inflight(lmdb_backend: LMDBBackend) -> None:
+    record = StoredRecord(source="s", record_type="t", body={}, sinks=["a", "b"])
+    lmdb_backend.put(record)
+    lmdb_backend.ack(record.record_id, sink="a")
+    lmdb_backend.dead(record.record_id, "boom", sink="b")
+    removed = lmdb_backend.delete_expired("2100-01-01T00:00:00+00:00", "2000-01-01T00:00:00+00:00")
+    assert removed == 0
+    removed = lmdb_backend.delete_expired("2100-01-01T00:00:00+00:00", "2100-01-01T00:00:00+00:00")
+    assert removed == 1
+    assert lmdb_backend.get(record.record_id) is None
+
+
+def test_migrates_legacy_record(tmp_db_path: Path) -> None:
+    import json
+
+    import lmdb
+
+    tmp_db_path.mkdir(parents=True)
+    env = lmdb.open(str(tmp_db_path), map_size=10_485_760, subdir=True)
+    payload = {
+        "record_id": "legacy",
+        "source": "s",
+        "record_type": "t",
+        "body": {"message": "old"},
+        "metadata": {},
+        "state": "failed",
+        "created_at": "2020-01-01T00:00:00+00:00",
+        "updated_at": "2020-01-01T00:00:00+00:00",
+        "retry_count": 2,
+        "next_retry_at": "2020-01-01T00:01:00+00:00",
+        "last_error": "boom",
+        "destination": "central",
+    }
+    with env.begin(write=True) as txn:
+        txn.put(b"rec:legacy", json.dumps(payload).encode())
+    env.close()
+
+    backend = LMDBBackend(path=str(tmp_db_path), map_size=10_485_760)
+    backend.open()
+    stored = backend.get("legacy")
+    assert stored is not None
+    assert stored.deliveries["central"].state == RecordState.failed
+    assert stored.deliveries["central"].retry_count == 2
+    assert stored.deliveries["central"].last_error == "boom"
+    assert backend.count(RecordState.failed) == 1
+    backend.close()

tests/test_plugins.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from daena.pipeline.processors.enricher import EnricherProcessor
+from daena.pipeline.processors.filter import FilterProcessor
+from daena.pipeline.record import Record
+from daena.pipeline.sinks.loki import LokiSink
+from daena.pipeline.sources.journald import (
+    JournaldSource,
+    build_journalctl_command,
+    entry_to_record,
+)
+
+
+def test_journalctl_command_tails_new_entries_without_cursor() -> None:
+    command = build_journalctl_command(cursor=None, units=["ssh.service"], matches=["_PID=1"])
+    assert command[:4] == ["journalctl", "--follow", "--output=json", "--no-pager"]
+    assert "--lines=0" in command
+    assert command[command.index("--unit") + 1] == "ssh.service"
+    assert "_PID=1" in command
+
+
+def test_journalctl_command_uses_cursor() -> None:
+    command = build_journalctl_command(cursor="abc", units=[], matches=[])
+    assert "--after-cursor=abc" in command
+    assert "--lines=0" not in command
+
+
+def test_entry_to_record_uses_journal_timestamp() -> None:
+    record = entry_to_record(
+        "system",
+        {
+            "MESSAGE": "hello",
+            "SYSLOG_IDENTIFIER": "sshd",
+            "_SYSTEMD_UNIT": "ssh.service",
+            "__CURSOR": "cur",
+            "__REALTIME_TIMESTAMP": "1600000000000000",
+        },
+    )
+    assert record.source == "system"
+    assert record.record_type == "journald"
+    assert record.body["message"] == "hello"
+    assert record.body["syslog_identifier"] == "sshd"
+    assert record.metadata["unit"] == "ssh.service"
+    assert record.timestamp.startswith("2020-")
+
+
+class _ListReader:
+    def __init__(self, entries: list[dict[str, Any]]) -> None:
+        self._entries = entries
+        self.stopped = False
+
+    async def run(self, on_entry) -> None:  # type: ignore[no-untyped-def]
+        for entry in self._entries:
+            await on_entry(entry)
+
+    async def stop(self) -> None:
+        self.stopped = True
+
+
+@pytest.mark.asyncio
+async def test_journald_source_emits_and_stores_cursor(tmp_path: Path) -> None:
+    cursor_path = tmp_path / "cursor"
+    reader = _ListReader(
+        [
+            {"MESSAGE": "one", "__CURSOR": "c1"},
+            {"MESSAGE": "two", "__CURSOR": "c2"},
+        ]
+    )
+    source = JournaldSource(
+        "journal",
+        config={"cursor_path": str(cursor_path)},
+        reader=reader,
+    )
+    seen: list[Record] = []
+
+    async def capture(record: Record) -> None:
+        seen.append(record)
+
+    source.set_callback(capture)
+    await source.start()
+    await asyncio.sleep(0.05)
+    await source.stop()
+    assert [record.body["message"] for record in seen] == ["one", "two"]
+    assert cursor_path.read_text(encoding="utf-8").strip() == "c2"
+    assert reader.stopped is True
+
+
+@pytest.mark.asyncio
+async def test_enricher_sets_hostname_and_static_fields() -> None:
+    processor = EnricherProcessor(
+        "host",
+        config={"hostname": True, "static": {"env": "prod"}, "target": "metadata"},
+    )
+    record = Record(source="s", record_type="t", body={"message": "m"})
+    result = await processor.process(record)
+    assert result is not None
+    assert result.metadata["hostname"]
+    assert result.metadata["env"] == "prod"
+
+
+@pytest.mark.asyncio
+async def test_filter_drop_and_keep() -> None:
+    dropper = FilterProcessor("drop", config={"drop": {"body.level": "debug"}})
+    debug = Record(source="s", record_type="t", body={"level": "debug"})
+    info = Record(source="s", record_type="t", body={"level": "info"})
+    assert await dropper.process(debug) is None
+    assert await dropper.process(info) is not None
+
+    keeper = FilterProcessor("keep", config={"keep": {"source": "journal"}})
+    assert await keeper.process(Record(source="other", record_type="t", body={})) is None
+    kept = await keeper.process(Record(source="journal", record_type="t", body={}))
+    assert kept is not None
+
+
+@pytest.mark.asyncio
+async def test_loki_sink_posts_stream() -> None:
+    response = MagicMock()
+    response.is_success = True
+    response.status_code = 204
+    client = AsyncMock()
+    client.post = AsyncMock(return_value=response)
+    client.aclose = AsyncMock()
+    with patch("daena.pipeline.sinks.loki.httpx.AsyncClient", return_value=client):
+        sink = LokiSink("loki", config={"url": "http://loki/push", "labels": {"job": "daena"}})
+        await sink.start()
+        record = Record(
+            source="system",
+            record_type="journald",
+            body={"message": "hello"},
+            timestamp="2020-09-13T12:26:40+00:00",
+        )
+        result = await sink.deliver(record)
+        await sink.stop()
+    assert result.success is True
+    payload = client.post.await_args.kwargs["json"]
+    stream = payload["streams"][0]
+    assert stream["stream"]["job"] == "daena"
+    assert stream["stream"]["source"] == "system"
+    assert stream["values"][0][1] == "hello"
+    assert stream["values"][0][0].isdigit()
+    json.dumps(payload)