diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index fa8abf1..fd81b49 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -27,6 +27,11 @@ import fluksio #: Where an installation keeps everything, unless it is told otherwise. DEFAULT_HOME = Path("~/.fluksio") +#: The portal a claim code is redeemed at, unless another is named. Having a +#: default is the difference between one flag and two on the one command that +#: is run before anything works. +DEFAULT_PORTAL = "https://hub.fluksio.com" + #: WAL — what lets readers work while the engine writes — is not supported on #: these. The database would be corrupt or locked, so it is worth saying. NETWORK_FILESYSTEMS = ("nfs", "nfs4", "cifs", "smb", "smb3", "lustre", "fuse.sshfs") @@ -233,13 +238,12 @@ def cmd_serve(args: argparse.Namespace) -> int: _print_new_admin(admin_email, generated) if args.enroll: - if not args.portal: - print("error: --enroll needs --portal", file=sys.stderr, flush=True) - return 1 if not cloud_config.exists(): # Before the engine starts, so the connector finds the config and # dials out as part of coming up rather than needing a restart. - failed = _enroll(args.portal, args.enroll, args.admin_email) + failed = _enroll( + args.portal or DEFAULT_PORTAL, args.enroll, args.admin_email + ) if failed: return failed @@ -274,7 +278,7 @@ def cmd_serve(args: argparse.Namespace) -> int: _say(" The dashboard is served by the portal; nothing is served here.") else: _say(" No portal. Pair this installation with:") - _say(" fluksio enroll --portal https://hub.example.com") + _say(" fluksio enroll ") _say(f" Signed in as {admin_email}") _say(f" token in {token_path}") _mention_other_installation(data_dir) @@ -328,14 +332,23 @@ def _parser() -> argparse.ArgumentParser: serve.add_argument("--admin-email", default=None) serve.add_argument("--admin-password", default=None) serve.add_argument("--enroll", metavar="CODE", help="claim code, if not yet paired") - serve.add_argument("--portal", metavar="URL", help="the portal --enroll redeems at") + serve.add_argument( + "--portal", + metavar="URL", + help=f"the portal --enroll redeems at (default {DEFAULT_PORTAL})", + ) serve.set_defaults(func=cmd_serve) enroll = subparsers.add_parser( "enroll", help="pair this installation with a portal" ) enroll.add_argument("code", help="the claim code minted on the portal") - enroll.add_argument("--portal", required=True, metavar="URL") + enroll.add_argument( + "--portal", + default=DEFAULT_PORTAL, + metavar="URL", + help=f"a portal of your own, instead of {DEFAULT_PORTAL}", + ) enroll.add_argument( "--as", dest="as_email", diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index 2152e19..60918aa 100644 --- a/backend/fluksio/cloud/connector.py +++ b/backend/fluksio/cloud/connector.py @@ -19,6 +19,7 @@ import asyncio import base64 import contextlib import logging +import random import time import uuid from dataclasses import replace @@ -36,7 +37,14 @@ logger = logging.getLogger(__name__) PROTOCOL = 1 HEARTBEAT_S = 20.0 STATUS_S = 60.0 +BASE_BACKOFF_S = 1.0 MAX_BACKOFF_S = 30.0 +#: How long a silent socket is given before it is treated as gone. A laptop +#: that suspended, or a wifi that changed underneath us, leaves a connection +#: the operating system still believes in — nothing arrives and nothing fails. +#: Only an unanswered ping tells us, so this is the worst case for noticing. +PING_INTERVAL_S = 20.0 +PING_TIMEOUT_S = 20.0 CHUNK_BYTES = 256 * 1024 MAX_WS_MESSAGE = 1024 * 1024 #: Only the versioned API is served over the tunnel. The MCP mount and the @@ -79,31 +87,53 @@ class CloudConnector: # ------------------------------------------------------------------------- async def serve_forever(self) -> None: - backoff = 1.0 + """Hold the link up, and put it back up when it goes down. + + Every ending is the same ending here: a portal that closed the socket, + a network that changed under it, a laptop that woke up somewhere else. + The link is dialled again in all of them — the only question is how + soon, and that turns on whether this attempt got as far as being + attached. One that did and then dropped is a network event, so it + retries at once; one that never stood up is met with a longer wait + each time, because whatever is refusing is unlikely to stop within a + second. + """ + delay = BASE_BACKOFF_S while True: config = cloud_config.load() if config is None: # Disconnected locally while we were running. return + attached = False try: - await self._session(config) - backoff = 1.0 + attached = await self._session(config) except asyncio.CancelledError: raise except Exception as exc: + self._last_error = str(exc) + logger.warning("Portal link down: %s", exc) + finally: self._connected = False self._connected_since = None - self._last_error = str(exc) - logger.warning("Portal link down: %s — retrying in %.0fs", exc, backoff) - await asyncio.sleep(backoff) - backoff = min(MAX_BACKOFF_S, backoff * 2) - async def _session(self, config: cloud_config.CloudConfig) -> None: + delay = BASE_BACKOFF_S if attached else min(MAX_BACKOFF_S, delay * 2) + # Jittered, so a portal coming back up is not met by every + # installation it serves in the same instant. + wait = delay * (0.75 + random.random() * 0.5) + logger.info("Reconnecting to the portal in %.0fs", wait) + await asyncio.sleep(wait) + + async def _session(self, config: cloud_config.CloudConfig) -> bool: + """One connection, from dial to close. True if it ever attached.""" import websockets + attached = False url = f"{config.ws_url}?token={config.token}" async with websockets.connect( - url, max_size=MAX_WS_MESSAGE, ping_interval=20 + url, + max_size=MAX_WS_MESSAGE, + ping_interval=PING_INTERVAL_S, + ping_timeout=PING_TIMEOUT_S, ) as socket: await socket.send( _dump({"op": "hello", "protocol": PROTOCOL, "app_version": _version()}) @@ -116,6 +146,7 @@ class CloudConnector: self._adopt_owner(config, welcome.get("owner")) + attached = True self._connected = True self._connected_since = time.time() self._last_error = None @@ -138,12 +169,18 @@ class CloudConnector: self._cancel_stream(str(frame.get("id") or "")) finally: keepalive.cancel() + # Awaited, not just cancelled: a keepalive that died of its own + # accord holds the reason the link went, and an un-awaited task + # takes it to the garbage collector instead of the log. + with contextlib.suppress(asyncio.CancelledError, Exception): + await keepalive self._connected = False self._connected_since = None for task in list(self._calls.values()) + list(self._streams.values()): task.cancel() self._calls.clear() self._streams.clear() + return attached @staticmethod def _adopt_owner(config: cloud_config.CloudConfig, owner: Any) -> None: diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 4af39fe..2ca5952 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -94,8 +94,21 @@ def discover(targets: list[str]) -> list[Flow]: for info in pkgutil.walk_packages(package.__path__, f"{dotted}."): importlib.import_module(info.name) continue + # A plain directory — a repository root, usually. Its own modules, + # and the packages inside it: `myresearch/` beside a `README` is the + # ordinary shape, and naming it explicitly should not be the price of + # keeping your code in a package. for module in sorted(path.glob("*.py")): _import(*_module_of(module)) + for child in sorted(path.iterdir()): + if child.name.startswith(".") or not (child / "__init__.py").exists(): + continue + root, dotted = _package_of(child) + _import(root, dotted) + for info in pkgutil.walk_packages( + sys.modules[dotted].__path__, f"{dotted}." + ): + importlib.import_module(info.name) return list(FLOWS.values()) @@ -125,8 +138,9 @@ def cmd_sync(args: argparse.Namespace) -> int: return _fail(str(exc)) if not flows: return _fail( - f"no flows declared in {', '.join(targets)} — a flow is a `flow(...)` " - "call at module level" + f"no flows declared in {', '.join(targets)} — a flow is a `Flow(...)` " + "at module level. Name the package if it is somewhere else: " + "`fluksio sync src/myresearch`." ) repo = repo_root(targets[0]) @@ -218,9 +232,37 @@ def _params(definition: dict[str, Any], rest: list[str]) -> dict[str, Any]: return params +def _sync_first(client: Client) -> None: + """Upload what the working directory declares, before running it. + + The reason a run exists is usually the edit that came before it, and + remembering to sync is remembering to do the thing the computer could have + done. So `run` syncs by default — including the worker refresh, which is + what makes an edit to your own package take effect at all. + + A directory that declares nothing is not an error: a flow drawn on the + canvas is run the same way, and has nothing to upload. + """ + try: + flows = discover(["."]) + except (ImportError, SyncError) as exc: + # Do not fail a run for a module the run may not even need. + _say(f"warning: nothing synced — {exc}") + return + if not flows: + return + repo = repo_root(".") + reports = sync(flows, client, origin=origin_of(repo)) + changed = [r for r in reports if not r.unchanged] + if changed: + _say(f"synced {', '.join(r.flow for r in changed)}") + + def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: try: client = Client(url=args.url, token=args.token) + if not args.no_sync: + _sync_first(client) stored = client.get_flow(args.flow) if stored is None: return _fail(f"no flow '{args.flow}' on that engine") @@ -301,12 +343,17 @@ def add_parsers(subparsers: Any) -> None: parser.set_defaults(func=cmd_sync) parser = subparsers.add_parser( - "run", help="start a run, passing the flow's inputs as --name value" + "run", help="sync this directory, then start a run of one of its flows" ) parser.add_argument("flow") parser.add_argument("--seed", type=int, default=None) parser.add_argument("--wait", action="store_true", help="block until it finishes") parser.add_argument("--timeout", type=float, default=0.0) + parser.add_argument( + "--no-sync", + action="store_true", + help="run what is already on the engine, without uploading first", + ) with_engine(parser) parser.set_defaults(func=cmd_run) diff --git a/backend/tests/sdk/test_paths.py b/backend/tests/sdk/test_paths.py index 602eaaa..852ecdc 100644 --- a/backend/tests/sdk/test_paths.py +++ b/backend/tests/sdk/test_paths.py @@ -89,3 +89,40 @@ def test_ignoring_itself_leaves_an_existing_gitignore_alone(elsewhere: Path): client.ignore_self(directory) assert (directory / ".gitignore").read_text() == "mine\n" + + +def test_a_repository_whose_code_is_in_a_package_is_discovered( + elsewhere: Path, monkeypatch: pytest.MonkeyPatch +): + """`fluksio sync` in a repo root has to find `myresearch/`, not just *.py. + + This is the ordinary layout, and it is what `run` leans on to sync without + being told where to look. + """ + from fluksio.sdk import FLOWS + from fluksio.sdk.cli import discover + + package = elsewhere / "mystudy" + package.mkdir() + (package / "__init__.py").write_text("") + (package / "pipeline.py").write_text( + "from fluksio import Flow, Port, node\n" + "\n" + "@node(provides=[Port('score', 'float')])\n" + "def scoring():\n" + " return {'score': 1.0}\n" + "\n" + "study = Flow('study', nodes=[scoring], outputs=['score'])\n" + ) + # The things a repository root also holds, none of which is importable. + (elsewhere / ".venv").mkdir() + (elsewhere / ".fluksio").mkdir() + (elsewhere / "data").mkdir() + + FLOWS.clear() + try: + found = discover(["."]) + finally: + FLOWS.clear() + + assert [flow.name for flow in found] == ["study"] diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 95e836b..77fab68 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -116,3 +116,24 @@ def test_serve_uses_the_installation_the_directory_belongs_to( # And `--data-dir` still names any directory outright. named = cli._data_dir(str(tmp_path / "named")) assert named == (tmp_path / "named").resolve() + + +def test_enrolling_needs_only_a_claim_code() -> None: + """The default portal is what makes first-run one flag rather than two.""" + from fluksio.cli import DEFAULT_PORTAL, _parser + + args = _parser().parse_args(["enroll", "ABC-123"]) + + assert args.portal == DEFAULT_PORTAL + assert DEFAULT_PORTAL.startswith("https://") + + # And a portal of your own still wins. + mine = _parser().parse_args(["enroll", "ABC-123", "--portal", "https://hub.me"]) + assert mine.portal == "https://hub.me" + + +def test_run_syncs_by_default_and_can_be_told_not_to() -> None: + from fluksio.cli import _parser + + assert _parser().parse_args(["run", "train"]).no_sync is False + assert _parser().parse_args(["run", "train", "--no-sync"]).no_sync is True diff --git a/backend/tests/test_reconnect.py b/backend/tests/test_reconnect.py new file mode 100644 index 0000000..7b3a81f --- /dev/null +++ b/backend/tests/test_reconnect.py @@ -0,0 +1,95 @@ +"""The portal link goes back up on its own, whatever took it down. + +A laptop that suspends, a wifi that changes, a portal that restarts: all of +them end the same session, and none of them should need anybody to notice. +""" + +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest + +from fluksio.cloud import config as cloud_config +from fluksio.cloud.connector import BASE_BACKOFF_S, MAX_BACKOFF_S, CloudConnector + + +class _Link: + """A connector whose sessions do whatever a test says, instantly.""" + + def __init__(self, connector: CloudConnector) -> None: + self.connector = connector + self.waits: list[float] = [] + self.sessions = 0 + + def run(self, outcomes: list[Any]) -> list[float]: + """Drive `serve_forever` through `outcomes`, then let it return.""" + + async def session(_config: Any) -> bool: + outcome = outcomes[self.sessions] + self.sessions += 1 + if self.sessions >= len(outcomes): + # The next iteration reads the config and stops. + cloud_config.load = lambda: None # type: ignore[assignment] + if isinstance(outcome, Exception): + raise outcome + return bool(outcome) + + async def sleep(seconds: float) -> None: + self.waits.append(seconds) + + self.connector._session = session # type: ignore[assignment] + with_patched = asyncio.sleep + asyncio.sleep = sleep # type: ignore[assignment] + try: + asyncio.run(self.connector.serve_forever()) + finally: + asyncio.sleep = with_patched # type: ignore[assignment] + return self.waits + + +@pytest.fixture +def link(monkeypatch: pytest.MonkeyPatch) -> _Link: + monkeypatch.setattr(cloud_config, "load", lambda: object()) + return _Link(CloudConnector(app=None)) # type: ignore[arg-type] + + +def test_a_portal_that_closes_the_socket_is_not_hammered(link: _Link): + """A clean close used to return normally and reconnect with no wait.""" + waits = link.run([True, True, True]) + + # One wait per session. Returning normally from a session used to skip the + # sleep entirely, so a portal that accepted and closed span the loop. + assert link.sessions == 3 + assert len(waits) == 3, waits + assert all(wait > 0 for wait in waits), waits + + +def test_a_link_that_stood_up_retries_at_once(link: _Link): + """A dropped connection is a network event, not a refusal.""" + waits = link.run([True, True]) + + # Jittered around the base delay rather than backed off. + assert len(waits) == 2, waits + assert all(BASE_BACKOFF_S * 0.7 <= w <= BASE_BACKOFF_S * 1.3 for w in waits), waits + + +def test_a_link_that_never_attaches_backs_off(link: _Link): + waits = link.run([OSError("no route to host")] * 6) + + assert waits[0] < waits[1] < waits[2], waits + assert max(waits) <= MAX_BACKOFF_S * 1.3 + + +def test_backing_off_stops_at_the_ceiling(link: _Link): + waits = link.run([OSError("refused")] * 12) + + assert max(waits) <= MAX_BACKOFF_S * 1.3, waits + + +def test_a_failure_after_a_good_session_starts_over_from_the_base(link: _Link): + """Ten minutes of uptime should not inherit the last outage's delay.""" + waits = link.run([OSError("x"), OSError("x"), OSError("x"), True, True]) + + assert waits[-1] <= BASE_BACKOFF_S * 1.3, waits diff --git a/docs/code/cli.md b/docs/code/cli.md index 4f595b7..fea966a 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -76,7 +76,7 @@ Created the admin account admin@example.com Fluksio 0.1.0 — data in /home/you/.fluksio API http://127.0.0.1:8000/api/v1 No portal. Pair this installation with: - fluksio enroll --portal https://hub.example.com + fluksio enroll ``` An enrolled installation says which portal it is on instead, and notes that the @@ -87,21 +87,25 @@ dashboard is served from there rather than here. Pairs an existing installation with a portal. ```sh -fluksio enroll ABCD-1234 --portal https://hub.fluksio.com +fluksio enroll ABCD-1234 ``` | Option | What it does | |---|---| -| `--portal URL` | **required** — the portal the code was minted on | +| `--portal URL` | a portal of your own, instead of `https://hub.fluksio.com` | | `--as EMAIL` | the local account a portal session arrives as | -| `--data-dir PATH` | which installation, if not the default | +| `--data-dir PATH` | which installation, if not the one this directory is in | Get the code from the portal under **Installations → Add installation**. It is single-use and expires in fifteen minutes. `--as` matters when the installation has several superusers — without it, enrolment refuses rather than guessing. -Afterwards, `fluksio serve` dials the portal as it comes up. See -[Accounts and the portal](../interface/portal.md). +Afterwards, `fluksio serve` dials the portal as it comes up, and keeps dialling: +a portal that restarts, a wifi that changes, a laptop that suspends and wakes +somewhere else all end the same connection, and the link is put back up without +anybody noticing. A connection that stood up and then dropped is retried at +once; one that never stood up waits a little longer each time, up to half a +minute. See [Accounts and the portal](../interface/portal.md). ## `fluksio worker` @@ -164,9 +168,15 @@ until the process goes. See fluksio run train --lr 0.05 --seed 7 [--wait] ``` -Submits a run. Flags that are not its own are the flow's inputs, typed by what -the flow declares them as. `--wait` blocks until the run finishes and exits -non-zero if it failed. +Syncs the working directory, then submits a run — so the command after an edit +is this one and nothing else. Flags that are not its own are the flow's +inputs, typed by what the flow declares them as. `--wait` blocks until the run +finishes and exits non-zero if it failed. + +`--no-sync` runs what is already on the engine. Worth it in a tight loop where +you know nothing changed, since syncing retires the workers and the next call +pays its imports again. A directory that declares no flows syncs nothing and +says nothing — a flow drawn on the canvas is run the same way. ### `fluksio runs` diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md index 65dbba1..126dcc7 100644 --- a/docs/getting-started/data-science.md +++ b/docs/getting-started/data-science.md @@ -26,7 +26,7 @@ Fluksio 0.1.0 — data in /home/you/my-research/.fluksio Nodes /home/you/my-research/.venv/bin/python your environment, adopted. Add packages with pip. No portal. Pair this installation with: - fluksio enroll --portal https://hub.example.com + fluksio enroll Signed in as admin@example.com token in /home/you/my-research/.fluksio/client.json ``` @@ -397,6 +397,11 @@ fluksio run train --lr 0.003 --seed 7 fluksio runs --flow train ``` +`run` syncs first, so after an edit the command is just `fluksio run` — there +is no step to forget. `--no-sync` skips it for a tight loop where nothing +changed, since syncing retires the workers and the next call pays its imports +again. + `--lr` is typed by the flow's own inputs, so `0.003` arrives as a float. A parameter you did not declare, or one of the wrong type, is refused before anything executes. It answers immediately with a queued run — training is