Reconcile rebuilds interrupted before replacement creation
This commit is contained in:
+73
-1
@@ -35,7 +35,7 @@ import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
VERSION = "1.8.3"
|
||||
VERSION = "1.8.4"
|
||||
LABEL = "io.electric-cherry"
|
||||
MAX_FILE = 32 * 1024 * 1024
|
||||
MANIFEST_MAX_ENTRIES = 10000
|
||||
@@ -2665,12 +2665,84 @@ class Agent:
|
||||
atomic_json(self.config_path, self.config)
|
||||
return True
|
||||
|
||||
def reconcile_uncreated_rebuild(self, operation, row):
|
||||
"""Settle removal only when the replacement create was never submitted.
|
||||
|
||||
The write lane and instance lock exclude the original Python execution.
|
||||
A terminal journal receipt prevents any later replay of that command.
|
||||
"""
|
||||
if row["state"] != "unknown":
|
||||
return row
|
||||
instance_id, generation = instance_identity(operation.get("instanceId"), operation.get("generation"))
|
||||
payload = operation.get("payload", {})
|
||||
next_generation = payload.get("nextGeneration", payload.get("newGeneration"))
|
||||
if type(next_generation) is not int or next_generation <= generation:
|
||||
return row
|
||||
previous = json.loads(row["result"]) if row.get("result") else {}
|
||||
if not isinstance(previous, dict):
|
||||
return row
|
||||
directory = scoped(self.root, f"instances/{instance_id}/{next_generation}")
|
||||
|
||||
def unchanged_generation():
|
||||
try:
|
||||
marker = scoped(self.root, f"instances/{instance_id}/current.json")
|
||||
current = json.loads(marker.read_text())
|
||||
except FileNotFoundError:
|
||||
return True
|
||||
return isinstance(current, dict) and type(current.get("generation")) is int and current["generation"] == generation and current.get("removed", False) is False
|
||||
|
||||
def both_missing():
|
||||
for target in (generation, next_generation):
|
||||
health = self.inspect(instance_id, target)
|
||||
if health.get("exists") is not False or health.get("running") is not False or health.get("state") != "missing":
|
||||
return False
|
||||
return True
|
||||
|
||||
if not unchanged_generation() or not both_missing():
|
||||
return row
|
||||
# Preparation can leave runtime/configuration files before create's
|
||||
# checkpoint. A fresh operation must not inherit those partial files.
|
||||
expected_spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration", "unhealthyRecovery")}
|
||||
try:
|
||||
saved_spec = json.loads(scoped(directory, "spec.json").read_text())
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
else:
|
||||
if saved_spec != expected_spec:
|
||||
return row
|
||||
try:
|
||||
metadata = directory.lstat()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
else:
|
||||
if not stat_module.S_ISDIR(metadata.st_mode):
|
||||
return row
|
||||
shutil.rmtree(directory)
|
||||
# Also sync when a previous reconciliation removed the directory but
|
||||
# crashed before recording the terminal receipt.
|
||||
try:
|
||||
sync_parent(directory)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
if not unchanged_generation() or not both_missing():
|
||||
return row
|
||||
original_error = row.get("error")
|
||||
result = {**previous, "reconciled": True, "originalError": original_error,
|
||||
"rebuildFailure": {"phase": "disrupting", "generation": generation, "nextGeneration": next_generation,
|
||||
"oldContainerMissing": True, "replacementContainerMissing": True, "replacementFilesRemoved": True}}
|
||||
message = "Old container removed; replacement creation failed."
|
||||
if original_error:
|
||||
message += " " + original_error
|
||||
return self.journal.update(operation["id"], "failed", result, error=message)
|
||||
|
||||
def reconcile(self, operation, row):
|
||||
"""Prove desired state with immutable ownership; never resubmit a command."""
|
||||
kind, payload = operation["kind"], operation.get("payload", {})
|
||||
instance_id, generation = operation.get("instanceId"), operation.get("generation")
|
||||
checkpoint = json.loads(row["checkpoint"]) if row.get("checkpoint") else {}
|
||||
try:
|
||||
if kind == "instance.rebuild" and checkpoint.get("phase") == "disrupting":
|
||||
return self.reconcile_uncreated_rebuild(operation, row)
|
||||
if kind == "instance.remove" and "autoscaling" in payload and not checkpoint:
|
||||
return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error="Interrupted autoscaling removal requires a new decision")
|
||||
if kind == "instance.cancel-create":
|
||||
|
||||
@@ -1604,6 +1604,174 @@ class AgentTests(unittest.TestCase):
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "completed")
|
||||
self.assertFalse(old.exists())
|
||||
|
||||
def interrupted_rebuild_before_create(self):
|
||||
payload = self.create()
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2}, "interrupted-rebuild")
|
||||
|
||||
def interrupted_seed(app, image):
|
||||
(app / "partial-runtime").write_text("incomplete")
|
||||
raise OSError(module.errno.ENOSPC, "disk full")
|
||||
|
||||
with patch.object(self.agent, "seed_plutonium", side_effect=interrupted_seed):
|
||||
row = self.agent.execute(operation)
|
||||
self.assertEqual(row["state"], "unknown", row)
|
||||
self.assertEqual(json.loads(row["checkpoint"])["phase"], "disrupting")
|
||||
self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"])
|
||||
self.assertFalse(self.agent.inspect(INSTANCE, 2)["exists"])
|
||||
return operation, row
|
||||
|
||||
def test_rebuild_reconciles_removal_before_create_without_replaying(self):
|
||||
operation, interrupted = self.interrupted_rebuild_before_create()
|
||||
original = self.agent.instance_dir(INSTANCE, 1)
|
||||
partial = self.agent.instance_dir(INSTANCE, 2)
|
||||
module.atomic_json(partial / "spec.json", operation["payload"]["spec"])
|
||||
self.agent.journal.close()
|
||||
self.agent = Agent(self.config, self.docker)
|
||||
calls = list(self.docker.calls)
|
||||
failed = self.agent.execute(operation)
|
||||
self.assertEqual(failed["state"], "failed", failed)
|
||||
result = json.loads(failed["result"])
|
||||
self.assertEqual(result["rebuildFailure"], {"phase": "disrupting", "generation": 1, "nextGeneration": 2,
|
||||
"oldContainerMissing": True, "replacementContainerMissing": True,
|
||||
"replacementFilesRemoved": True})
|
||||
self.assertTrue(result["reconciled"])
|
||||
self.assertEqual(result["unexpectedError"], json.loads(interrupted["result"])["unexpectedError"])
|
||||
self.assertEqual(result["originalError"], interrupted["error"])
|
||||
self.assertTrue(failed["error"].startswith("Old container removed; replacement creation failed."))
|
||||
self.assertIn(interrupted["error"], failed["error"])
|
||||
self.assertEqual((original / "app/configs/server.cfg").read_bytes(), b"original")
|
||||
self.assertFalse(partial.exists())
|
||||
self.assertEqual(json.loads((original.parent / "current.json").read_text()), {"generation": 1})
|
||||
self.assertEqual(self.agent.execute(operation), failed)
|
||||
self.assertEqual(self.docker.calls, calls)
|
||||
replacement = self.operation("instance.rebuild", operation["payload"], "replacement-rebuild")
|
||||
self.assertEqual(self.agent.execute(replacement)["state"], "completed")
|
||||
self.assertFalse((partial / "app/partial-runtime").exists())
|
||||
self.assertEqual(self.agent.inspect(INSTANCE, 2)["creationOperation"], replacement["id"])
|
||||
calls = list(self.docker.calls)
|
||||
self.agent.journal.close()
|
||||
self.agent = Agent(self.config, self.docker)
|
||||
self.assertEqual(self.agent.execute(operation), failed)
|
||||
self.assertEqual(self.docker.calls, calls)
|
||||
|
||||
def test_rebuild_removal_reconciles_when_marker_is_absent(self):
|
||||
operation, _ = self.interrupted_rebuild_before_create()
|
||||
marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json"
|
||||
marker.unlink()
|
||||
row = self.agent.execute(operation)
|
||||
self.assertEqual(row["state"], "failed", row)
|
||||
self.assertIn("rebuildFailure", json.loads(row["result"]))
|
||||
self.assertFalse(marker.exists())
|
||||
self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists())
|
||||
|
||||
def test_rebuild_lost_removal_receipt_settles_without_recreating_or_deleting_twice(self):
|
||||
payload = self.create()
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
self.docker.lose_reply_after = "rm"
|
||||
operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2}, "rebuild")
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "unknown")
|
||||
self.assertFalse(self.agent.instance_dir(INSTANCE, 2).exists())
|
||||
calls = list(self.docker.calls)
|
||||
result = self.agent.execute(operation)
|
||||
self.assertEqual(result["state"], "failed", result)
|
||||
self.assertTrue(json.loads(result["result"])["rebuildFailure"]["replacementFilesRemoved"])
|
||||
self.assertEqual(self.docker.calls, calls)
|
||||
self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists())
|
||||
|
||||
def test_rebuild_removal_requires_both_containers_positively_missing(self):
|
||||
operation, interrupted = self.interrupted_rebuild_before_create()
|
||||
partial = self.agent.instance_dir(INSTANCE, 2)
|
||||
for generation in (1, 2):
|
||||
for status, foreign in (("running", False), ("created", False), ("exited", False), ("unknown", False), ("running", True)):
|
||||
with self.subTest(generation=generation, status=status, foreign=foreign):
|
||||
name = self.agent.container_name(INSTANCE, generation)
|
||||
labels = {} if foreign else {module.LABEL + ".managed": "true", module.LABEL + ".instance": INSTANCE,
|
||||
module.LABEL + ".generation": str(generation)}
|
||||
self.docker.containers[name] = {"Config": {"Labels": labels}, "State": {"Status": status, "Running": status == "running"}}
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertTrue(partial.exists())
|
||||
self.docker.containers.pop(name)
|
||||
with patch.object(self.agent, "inspect", return_value={"exists": False, "running": False, "state": "unknown"}):
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
with patch.object(self.agent, "inspect", side_effect=AgentError("Docker unavailable")):
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertTrue(partial.exists())
|
||||
|
||||
def test_rebuild_removal_does_not_settle_after_a_create_checkpoint(self):
|
||||
operation, interrupted = self.interrupted_rebuild_before_create()
|
||||
for phase in ("creating", "starting", "unknown"):
|
||||
with self.subTest(phase=phase):
|
||||
self.agent.journal.checkpoint(operation["id"], {"phase": phase, "generation": 2})
|
||||
row = self.agent.execute(operation)
|
||||
self.assertEqual(row["state"], "unknown", row)
|
||||
self.assertEqual(row["result"], interrupted["result"])
|
||||
self.assertTrue(self.agent.instance_dir(INSTANCE, 2).exists())
|
||||
self.agent.journal.db.execute("UPDATE operations SET checkpoint=NULL WHERE id=?", (operation["id"],))
|
||||
self.agent.journal.db.commit()
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "unknown")
|
||||
self.assertTrue(self.agent.instance_dir(INSTANCE, 2).exists())
|
||||
|
||||
def test_rebuild_removal_rejects_ambiguous_generation_markers(self):
|
||||
operation, interrupted = self.interrupted_rebuild_before_create()
|
||||
marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json"
|
||||
for current in ({"generation": 0}, {"generation": 2}, {"generation": 3}, {"generation": True}, {"generation": "1"},
|
||||
{"generation": 1, "removed": True}, {"generation": 1, "removed": None}, {}, []):
|
||||
with self.subTest(current=current):
|
||||
module.atomic_json(marker, current)
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertTrue(self.agent.instance_dir(INSTANCE, 2).exists())
|
||||
self.assertEqual(json.loads(marker.read_text()), current)
|
||||
marker.write_text("invalid JSON")
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
|
||||
def test_rebuild_removal_keeps_unknown_for_unrelated_partial_files_and_symlinks(self):
|
||||
operation, interrupted = self.interrupted_rebuild_before_create()
|
||||
partial = self.agent.instance_dir(INSTANCE, 2)
|
||||
module.atomic_json(partial / "spec.json", {**operation["payload"]["spec"], "name": "unrelated"})
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertTrue(partial.exists())
|
||||
(partial / "spec.json").unlink()
|
||||
outside = self.agent.root / "retained-files"
|
||||
partial.rename(outside)
|
||||
partial.symlink_to(outside, target_is_directory=True)
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertEqual((outside / "app/partial-runtime").read_text(), "incomplete")
|
||||
partial.unlink()
|
||||
outside.rename(partial)
|
||||
marker = partial.parent / "current.json"
|
||||
retained_marker = self.agent.root / "retained-current.json"
|
||||
marker.rename(retained_marker)
|
||||
marker.symlink_to(retained_marker)
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertTrue(partial.exists())
|
||||
|
||||
def test_rebuild_removal_keeps_cleanup_errors_and_changed_generation_unresolved(self):
|
||||
operation, interrupted = self.interrupted_rebuild_before_create()
|
||||
partial = self.agent.instance_dir(INSTANCE, 2)
|
||||
with patch.object(module.shutil, "rmtree", side_effect=PermissionError("cleanup denied")):
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertTrue(partial.exists())
|
||||
cleanup = module.shutil.rmtree
|
||||
|
||||
def superseded(path):
|
||||
cleanup(path)
|
||||
module.atomic_json(partial.parent / "current.json", {"generation": 3})
|
||||
|
||||
with patch.object(module.shutil, "rmtree", side_effect=superseded):
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertEqual(json.loads((partial.parent / "current.json").read_text()), {"generation": 3})
|
||||
self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists())
|
||||
|
||||
def test_rebuild_removal_waits_for_durable_partial_file_cleanup(self):
|
||||
operation, interrupted = self.interrupted_rebuild_before_create()
|
||||
with patch.object(module, "sync_parent", side_effect=OSError("sync unavailable")):
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertFalse(self.agent.instance_dir(INSTANCE, 2).exists())
|
||||
self.assertEqual(self.agent.execute(operation), interrupted)
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "failed")
|
||||
self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists())
|
||||
|
||||
def test_traversal_and_symlinks_are_rejected(self):
|
||||
self.create()
|
||||
app = self.agent.instance_dir(INSTANCE, 1) / "app"
|
||||
|
||||
Reference in New Issue
Block a user