diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a17dc8..9f2e4ad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,15 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Added + +- Contracts may set `timeout_minutes` on any node: an integer from 1 to 10,080 + that is the node's task timeout. Other values are refused as a contract + error naming the node. The merged wire entry carries the key only when set, + and it is the one wire key outside `config_hash`: changing it alone leaves + `config_hash` and `content_hash` unchanged. A node that does not set it has + the wire entry it always had. + ### Changed - The "mode_type not allowed for kind" contract error now quotes the offending diff --git a/README.md b/README.md index bb1e228..586ceb4 100644 --- a/README.md +++ b/README.md @@ -176,6 +176,15 @@ produced, never how it is checked or written. script reads with `os.environ[...]`. The contract never holds a secret value or env var name; the Secret itself is created in the cluster by the operator. +### Task timeout + +Any node may set `timeout_minutes:` — an integer from 1 to 10,080 (one week) — +to declare how long its task may run before continuo gives up on it. A value +outside that range, or one that is not an integer, fails +`continuo-runtime validate`/`merge`/`lint`. It is the one key that is not part +of `config_hash`, so changing only the timeout never changes the node's +`content_hash`. + ## Write modes `mode_type:` chooses how the harness writes a node's table on each run: diff --git a/continuo_python_runtime/contract/loader.py b/continuo_python_runtime/contract/loader.py index 911d515..61c150d 100644 --- a/continuo_python_runtime/contract/loader.py +++ b/continuo_python_runtime/contract/loader.py @@ -44,8 +44,13 @@ "secret_ref", "mode_type", "primary_key", + "timeout_minutes", } +# Bounds of a node's declared task timeout, in minutes (one minute to one week). +MIN_TIMEOUT_MINUTES = 1 +MAX_TIMEOUT_MINUTES = 10_080 + _REQUIRED_STRING_FIELDS = ("schema", "table", "owner", "schedule") _ALLOWED_OUTPUT_COLUMN_KEYS = {"name", "type", "nullable"} @@ -295,6 +300,16 @@ def parse_node( raise ContractError(f"{label}: 'secret_ref' is not allowed for kind {kind}") secret_ref = validate_secret_ref(raw["secret_ref"], label) + timeout_minutes = raw.get("timeout_minutes") + if "timeout_minutes" in raw and ( + type(timeout_minutes) is not int + or not MIN_TIMEOUT_MINUTES <= timeout_minutes <= MAX_TIMEOUT_MINUTES + ): + raise ContractError( + f"{label}: 'timeout_minutes' must be an integer between " + f"{MIN_TIMEOUT_MINUTES} and {MAX_TIMEOUT_MINUTES}, got {timeout_minutes!r}" + ) + mode_type = raw.get("mode_type", DEFAULT_MODE_TYPE) if not isinstance(mode_type, str) or mode_type not in MODE_TYPES: raise ContractError( @@ -375,6 +390,7 @@ def parse_node( secret_ref=secret_ref, mode_type=mode_type, primary_key=primary_key, + timeout_minutes=timeout_minutes, ) diff --git a/continuo_python_runtime/contract/merge.py b/continuo_python_runtime/contract/merge.py index 7e9ec66..e656b57 100644 --- a/continuo_python_runtime/contract/merge.py +++ b/continuo_python_runtime/contract/merge.py @@ -170,6 +170,10 @@ def build_wire_contract( primary_bytes, member_bytes = _HASH_INPUTS[node.kind](node, repo_root) entry.update(hash_parts(entry, primary_bytes, member_bytes)) + # The one wire key outside config_hash: it is written after the hashes + # are computed, so changing it alone never changes content_hash. + if node.timeout_minutes is not None: + entry["timeout_minutes"] = node.timeout_minutes wire_nodes.append(entry) diff --git a/continuo_python_runtime/contract/model.py b/continuo_python_runtime/contract/model.py index bb9097b..39c0cd6 100644 --- a/continuo_python_runtime/contract/model.py +++ b/continuo_python_runtime/contract/model.py @@ -45,6 +45,7 @@ class Node: secret_ref: str = "" mode_type: str = DEFAULT_MODE_TYPE primary_key: tuple[str, ...] = () + timeout_minutes: int | None = None @property def relation(self) -> str: diff --git a/continuo_python_runtime/hashing.py b/continuo_python_runtime/hashing.py index bda3d15..3229fab 100644 --- a/continuo_python_runtime/hashing.py +++ b/continuo_python_runtime/hashing.py @@ -21,6 +21,10 @@ HASH_PART_FIELDS = ("source_hash", "shared_code_hash", "config_hash") CONTENT_HASH_FIELD = "content_hash" +# A wire key that is not part of the config_hash basis: it tunes how the node +# runs (the task timeout) and says nothing about what the node computes, so +# changing it alone is not a code change. +NON_HASHED_FIELDS = ("timeout_minutes",) def canonical_entry(entry: dict) -> dict: @@ -28,13 +32,14 @@ def canonical_entry(entry: dict) -> dict: Drops ``content_hash`` and all three of ``HASH_PART_FIELDS`` from the basis, and whitespace-normalizes every ``str`` value in ``reads`` via - ``" ".join(sql.split())``. Everything else is preserved verbatim, - including ``config``. + ``" ".join(sql.split())``. ``timeout_minutes`` is the one other wire key + left out of the basis (``NON_HASHED_FIELDS``). Everything else is + preserved verbatim, including ``config``. """ canonical = deepcopy(entry) canonical.pop(CONTENT_HASH_FIELD, None) - for field in HASH_PART_FIELDS: + for field in (*HASH_PART_FIELDS, *NON_HASHED_FIELDS): canonical.pop(field, None) reads = canonical.get("reads") diff --git a/docs/boundary-contract.md b/docs/boundary-contract.md index 5501bf8..cce1fd9 100644 --- a/docs/boundary-contract.md +++ b/docs/boundary-contract.md @@ -61,6 +61,16 @@ s3://///contract.yaml node's wire entry or `config_hash` changes), and it is part of `config_hash`, so `content_hash` changes when the referenced Secret name changes. `source_hash` does not. +- **`timeout_minutes`** is additive and optional, valid on every `kind`: an + integer from 1 to 10,080 (one week) that is the node's task timeout. Any + other value (zero, above the bound, a string, a float, a boolean) fails + `ContractError` naming the node and the key. On the merged wire entry it + appears only when the node declares it, and it is the one wire key **outside + `config_hash`**: it is written after the hash parts are computed and is + excluded from the `config_hash` basis, so changing it alone changes none of + the four hashes and is not a code change. A node that does not declare it + has the wire entry, `config_hash` and `content_hash` it always had, and + `contract_version` stays 1. - **`mode_type`** and **`primary_key`** are additive and optional. `mode_type` (`replace`, the default; `append`; `upsert`) says how the harness writes the node's table on each run; `python-csv` allows only `replace`. @@ -222,7 +232,8 @@ content_hash = "sha256:" + sha256(source_hash + "|" + shared_code_hash collapse to a single space, then stripped); `content_hash` **and all three part fields** (`source_hash`, `shared_code_hash`, `config_hash`) are excluded from the basis — `config_hash` cannot include itself, and - none of the four hash fields participates in another's basis. Every + none of the four hash fields participates in another's basis. So is + `timeout_minutes`, the one non-hash wire key left out of the basis. Every other field, including `config`, participates verbatim. Invariant: formatting-only edits (spaces in SQL, yaml indentation/key diff --git a/tests/contract/test_loader.py b/tests/contract/test_loader.py index f356747..ac55241 100644 --- a/tests/contract/test_loader.py +++ b/tests/contract/test_loader.py @@ -878,3 +878,22 @@ def test_primary_key_names_a_column_that_differs_only_in_case_from_another(): entry["output_columns"].append({"name": "Base", "type": "VARCHAR(3)", "nullable": False}) node = parse_node(entry, "c.yml") assert node.primary_key == ("rate_date", "Base") + + +# --- timeout_minutes ------------------------------------------------------------ + + +def test_timeout_minutes_is_read(): + assert parse_node({**VALID, "timeout_minutes": 120}, "f.yml").timeout_minutes == 120 + assert parse_node({**VALID, "timeout_minutes": 1}, "f.yml").timeout_minutes == 1 + assert parse_node({**VALID, "timeout_minutes": 10080}, "f.yml").timeout_minutes == 10080 + + +def test_timeout_minutes_defaults_to_none(): + assert parse_node(VALID, "f.yml").timeout_minutes is None + + +@pytest.mark.parametrize("bad", [0, 10081, "2h", 1.5, True]) +def test_a_bad_timeout_minutes_is_refused(bad): + with pytest.raises(ContractError, match=r"f\.yml \(analytics\.t\): 'timeout_minutes'"): + parse_node({**VALID, "timeout_minutes": bad}, "f.yml") diff --git a/tests/contract/test_merge.py b/tests/contract/test_merge.py index e161770..05c81af 100644 --- a/tests/contract/test_merge.py +++ b/tests/contract/test_merge.py @@ -617,3 +617,25 @@ def test_a_replace_entry_hashes_exactly_as_it_did_before_write_modes(extra): entry = node_entry(parse_node({**_REPLACE_NODE, **extra}, "f.yml")) assert "mode_type" not in entry and "primary_key" not in entry assert config_hash(entry) == _PRE_WRITE_MODES_CONFIG_HASH + + +# --- timeout_minutes ---------------------------------------------------------- + + +def test_timeout_minutes_is_outside_the_config_hash(tmp_path): + without = _only_entry(_modes_repo(tmp_path / "a")) + short = _only_entry(_modes_repo(tmp_path / "b", timeout_minutes=120)) + long = _only_entry(_modes_repo(tmp_path / "c", timeout_minutes=480)) + assert short["config_hash"] == long["config_hash"] == without["config_hash"] + assert short["content_hash"] == long["content_hash"] == without["content_hash"] + # The wire entry hashes exactly as the same entry would without the key. + assert config_hash(short) == short["config_hash"] + + +def test_the_wire_entry_carries_timeout_minutes_only_when_set(tmp_path): + without = _only_entry(_modes_repo(tmp_path / "a")) + declared = _only_entry(_modes_repo(tmp_path / "b", timeout_minutes=120)) + assert "timeout_minutes" not in without + assert set(without) == WIRE_ENTRY_KEYS + assert declared["timeout_minutes"] == 120 + assert set(declared) == WIRE_ENTRY_KEYS | {"timeout_minutes"} diff --git a/tests/test_hashing.py b/tests/test_hashing.py index 425f331..6df6014 100644 --- a/tests/test_hashing.py +++ b/tests/test_hashing.py @@ -186,3 +186,10 @@ def test_hash_parts_returns_four_keys_and_consistent_fold(): assert parts["content_hash"] == content_hash_fold( parts["source_hash"], parts["shared_code_hash"], parts["config_hash"] ) + + +def test_timeout_minutes_is_outside_the_config_hash_basis(): + with_timeout = {**ENTRY, "timeout_minutes": 120} + assert "timeout_minutes" not in canonical_entry(with_timeout) + assert config_hash(with_timeout) == config_hash(ENTRY) + assert config_hash({**ENTRY, "timeout_minutes": 480}) == config_hash(ENTRY)