Skip to content
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).

## [Unreleased]

### Changed
- Keep save-query retry policy internal and retry transient misses every 10 server ticks instead of exposing a retry-count setting.

## [0.1.3] - 2026-10-06

### Fixed
Expand Down
6 changes: 2 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,6 @@ The plugin writes `plugins/endkeep/config.toml` on first start.
enabled = true

[capture]
query_retries = 300
times = [
"12:00",
"16:30",
Expand Down Expand Up @@ -108,9 +107,8 @@ path = "backups"
min_free_space_gib = 5
```

`capture.query_retries` is the number of additional `save query` attempts allowed after the first unsuccessful
attempt. The default is 300; `0` disables additional retries. The existing 15-second query deadline remains a
hard safety cap. Older configuration files that omit this field automatically use 300.
Transient `save query` misses are retried internally every 10 server ticks. The capture still has a fixed
15-second query deadline, after which EndKeep enters the fail-safe `save resume` recovery path.

Retention keeps a snapshot when it is **within `keep_days` OR among the last `keep_last`**.
The raw hard limits are safety limits: if the queue cannot be logicalized and a hard limit must be
Expand Down
7 changes: 1 addition & 6 deletions src/endstone_endkeep/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,6 @@ def _section(mapping: Mapping[str, Any], name: str) -> Mapping[str, Any]:
@dataclass(frozen=True)
class CaptureConfig:
times: tuple[str, ...]
query_retries: int


@dataclass(frozen=True)
Expand Down Expand Up @@ -115,10 +114,6 @@ def from_mapping(cls, mapping: Mapping[str, Any]) -> EndKeepConfig:
retention = _section(mapping, "retention")
storage = _section(mapping, "storage")

query_retries = capture.get("query_retries", 300)
if not isinstance(query_retries, int) or isinstance(query_retries, bool) or query_retries < 0:
raise ConfigError("capture.query_retries must be a non-negative integer")

storage_path = storage.get("path")
if not isinstance(storage_path, str) or not storage_path.strip():
raise ConfigError("storage.path must be a non-empty path string")
Expand All @@ -138,7 +133,7 @@ def from_mapping(cls, mapping: Mapping[str, Any]) -> EndKeepConfig:

return cls(
enabled=enabled,
capture=CaptureConfig(times=_string_list(capture, "times"), query_retries=query_retries),
capture=CaptureConfig(_string_list(capture, "times")),
maintenance=MaintenanceConfig(_string_list(maintenance, "times")),
raw=RawConfig(
max_pending=_positive_int(raw, "max_pending"),
Expand Down
1 change: 0 additions & 1 deletion src/endstone_endkeep/config.toml
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
enabled = true

[capture]
query_retries = 300
times = [
"12:00",
"16:30",
Expand Down
27 changes: 12 additions & 15 deletions src/endstone_endkeep/coordinator.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,29 +30,27 @@ class CaptureCoordinator:
"""Main-thread BDS state machine with exact staging/durability on one worker."""

QUERY_TIMEOUT_SECONDS = 15.0
QUERY_POLL_SECONDS = 0.05
QUERY_RETRY_TICKS = 10

def __init__(
self,
plugin: Plugin,
raw_store: RawSnapshotStore,
*,
query_retries: int = 300,
space_guard: Callable[[int], bool] | None = None,
space_recovery: Callable[[int, str | None], bool] | None = None,
) -> None:
self._plugin = plugin
self._adapter = BdsSaveAdapter(plugin.server)
self._raw_store = raw_store
self._query_retries = query_retries
self._query_failures = 0
self._query_retry_ticks = 0
self._space_guard = space_guard
self._space_recovery = space_recovery
self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="endkeep-capture")
self._state: CaptureState = "idle"
self._scheduled_for: str | None = None
self._deadline = 0.0
self._next_query = 0.0
self._hold_started = 0.0
self._hold_elapsed = 0.0
self._cancel = Event()
Expand Down Expand Up @@ -88,6 +86,7 @@ def start_capture(self, *, scheduled_for: str | None) -> bool:
self._future = None
self._space_retry_bytes = None
self._query_failures = 0
self._query_retry_ticks = 0

hold_started = time.monotonic()
try:
Expand All @@ -100,7 +99,6 @@ def start_capture(self, *, scheduled_for: str | None) -> bool:
now = time.monotonic()
self._hold_started = hold_started
self._deadline = now + self.QUERY_TIMEOUT_SECONDS
self._next_query = now
self._state = "querying"
return True

Expand Down Expand Up @@ -136,14 +134,15 @@ def _pump_query(self) -> None:
if now >= self._deadline:
self._fail_held(f"save query timed out after {self._query_failures} unsuccessful attempt(s)")
return
if now < self._next_query:
return
self._next_query = now + self.QUERY_POLL_SECONDS
if self._query_retry_ticks > 0:
self._query_retry_ticks -= 1
if self._query_retry_ticks > 0:
return

try:
capture = self._adapter.query()
except Exception as exc:
self._retry_query_or_fail(f"save query raised: {exc}")
self._schedule_query_retry(f"save query raised: {exc}")
return

try:
Expand All @@ -152,7 +151,7 @@ def _pump_query(self) -> None:
# Endstone's dispatch_command() return value reflects command success,
# not whether the vanilla command reached BDS. A false result is expected
# while save query is waiting for the held save to become ready.
self._retry_query_or_fail("save query did not return a valid manifest")
self._schedule_query_retry("save query did not return a valid manifest")
return

if self._space_guard is not None and not self._space_guard(manifest.total_bytes):
Expand All @@ -168,11 +167,9 @@ def _pump_query(self) -> None:
)
self._state = "staging"

def _retry_query_or_fail(self, reason: str) -> None:
def _schedule_query_retry(self, _reason: str) -> None:
self._query_failures += 1
if self._query_failures <= self._query_retries:
return
self._fail_held(f"{reason}; retry limit exhausted after {self._query_failures} unsuccessful attempt(s)")
self._query_retry_ticks = self.QUERY_RETRY_TICKS

def _defer_for_space(self, required_bytes: int) -> None:
self._space_retry_bytes = required_bytes
Expand Down Expand Up @@ -290,11 +287,11 @@ def _reset(self) -> None:
self._state = "idle"
self._scheduled_for = None
self._deadline = 0.0
self._next_query = 0.0
self._hold_started = 0.0
self._hold_elapsed = 0.0
self._future = None
self._manifest = None
self._staged = None
self._space_retry_bytes = None
self._query_failures = 0
self._query_retry_ticks = 0
1 change: 0 additions & 1 deletion src/endstone_endkeep/plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,6 @@ def _configure_runtime(self, runtime_config: EndKeepConfig | None = None) -> Non
capture = CaptureCoordinator(
self,
raw_store,
query_retries=config.capture.query_retries,
space_guard=repository.queue.free_space_allows,
space_recovery=self._request_capture_space_recovery,
)
Expand Down
32 changes: 21 additions & 11 deletions tests/test_capture_coordinator.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ def __init__(
self.query_messages = query_messages
self.query_dispatched = query_dispatched
self.resume_failures = resume_failures
self.query_calls = 0
self.resume_calls = 0

@property
Expand All @@ -58,6 +59,7 @@ def hold(self):
return SimpleNamespace(dispatched=True, messages=(), errors=())

def query(self):
self.query_calls += 1
if self.query_error is not None:
raise self.query_error
return SimpleNamespace(dispatched=self.query_dispatched, messages=self.query_messages, errors=())
Expand Down Expand Up @@ -91,22 +93,25 @@ def _coordinator(
monkeypatch,
tmp_path: Path,
adapter: _FakeAdapter,
*,
query_retries: int = 300,
) -> tuple[CaptureCoordinator, _Plugin, _RawStore]:
plugin = _Plugin()
raw_store = _RawStore(tmp_path / "raw" / "snapshot")
monkeypatch.setattr(coordinator_module, "BdsSaveAdapter", lambda _server: adapter)
coordinator = CaptureCoordinator(plugin, raw_store, query_retries=query_retries)
coordinator = CaptureCoordinator(plugin, raw_store)
return coordinator, plugin, raw_store


def test_held_failure_retries_resume_and_blocks_new_capture(monkeypatch, tmp_path: Path) -> None:
adapter = _FakeAdapter(query_error=RuntimeError("query failed"), resume_failures=1)
coordinator, plugin, _raw_store = _coordinator(monkeypatch, tmp_path, adapter, query_retries=0)
coordinator, plugin, _raw_store = _coordinator(monkeypatch, tmp_path, adapter)
try:
assert coordinator.start_capture(scheduled_for=None)

coordinator.pump()
assert coordinator.status().state == "querying"
assert coordinator.status().held

coordinator._deadline = 0.0
coordinator.pump()

assert coordinator.status().state == "resume_retry"
Expand Down Expand Up @@ -196,7 +201,7 @@ def test_space_shortage_resumes_and_requests_cleanup_before_fresh_capture(monkey

def test_false_dispatch_result_is_retried_instead_of_failing_capture(monkeypatch, tmp_path: Path) -> None:
adapter = _FakeAdapter(query_dispatched=False)
coordinator, _plugin, _raw_store = _coordinator(monkeypatch, tmp_path, adapter, query_retries=1)
coordinator, _plugin, _raw_store = _coordinator(monkeypatch, tmp_path, adapter)
try:
assert coordinator.start_capture(scheduled_for="20:30")

Expand All @@ -206,12 +211,17 @@ def test_false_dispatch_result_is_retried_instead_of_failing_capture(monkeypatch
assert coordinator.status().held
assert adapter.resume_calls == 0

coordinator._next_query = 0.0
coordinator.pump()
assert adapter.query_calls == 1

assert coordinator.status().state == "idle"
assert not coordinator.status().held
assert adapter.resume_calls == 1
for _ in range(9):
coordinator.pump()
assert adapter.query_calls == 1

coordinator.pump()
assert adapter.query_calls == 2
assert coordinator.status().state == "querying"
assert coordinator.status().held
assert adapter.resume_calls == 0
finally:
coordinator.close()

Expand All @@ -221,7 +231,7 @@ def test_valid_manifest_is_accepted_even_when_dispatch_returns_false(monkeypatch
query_dispatched=False,
query_messages=("level/level.dat:3",),
)
coordinator, _plugin, _raw_store = _coordinator(monkeypatch, tmp_path, adapter, query_retries=0)
coordinator, _plugin, _raw_store = _coordinator(monkeypatch, tmp_path, adapter)
try:
assert coordinator.start_capture(scheduled_for="20:35")

Expand Down
21 changes: 0 additions & 21 deletions tests/test_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ def test_default_config() -> None:
config = EndKeepConfig.from_mapping(_defaults())
assert config.enabled is True
assert config.capture.times == ("12:00", "16:30", "20:30", "23:45")
assert config.capture.query_retries == 300
assert config.maintenance.times == ("06:00", "18:30")
assert config.maintenance.mode_for("06:00") == "FULL"
assert config.maintenance.mode_for("18:30") == "LOGIC_ONLY"
Expand All @@ -36,23 +35,3 @@ def test_default_config() -> None:
def test_invalid_time(value: str) -> None:
with pytest.raises(ConfigError):
parse_hhmm(value)


@pytest.mark.parametrize("value", [-1, 1.5, True, "3"])
def test_invalid_query_retries(value: object) -> None:
mapping = _defaults()
mapping["capture"]["query_retries"] = value
with pytest.raises(ConfigError):
EndKeepConfig.from_mapping(mapping)


def test_zero_query_retries_is_allowed() -> None:
mapping = _defaults()
mapping["capture"]["query_retries"] = 0
assert EndKeepConfig.from_mapping(mapping).capture.query_retries == 0


def test_missing_query_retries_uses_backward_compatible_default() -> None:
mapping = _defaults()
del mapping["capture"]["query_retries"]
assert EndKeepConfig.from_mapping(mapping).capture.query_retries == 300