diff --git a/.github/scripts/check_tool_versions.py b/.github/scripts/check_tool_versions.py index 9d0a8f4989..03a7510b62 100755 --- a/.github/scripts/check_tool_versions.py +++ b/.github/scripts/check_tool_versions.py @@ -1,6 +1,6 @@ #!/usr/bin/env -S uv run --script # /// script -# requires-python = ">=3.10" +# requires-python = ">=3.11" # dependencies = ["requests==2.34.2"] # /// diff --git a/.github/scripts/check_tool_versions.py.lock b/.github/scripts/check_tool_versions.py.lock index a8946713cd..d3fd8dc8eb 100644 --- a/.github/scripts/check_tool_versions.py.lock +++ b/.github/scripts/check_tool_versions.py.lock @@ -1,6 +1,6 @@ version = 1 -revision = 3 -requires-python = ">=3.10" +revision = 5 +requires-python = ">=3.11" [manifest] requirements = [{ name = "requests", specifier = "==2.34.2" }] @@ -20,22 +20,6 @@ version = "3.5.1" source = { registry = "https://pypi.org/simple" } sdist = { url = "https://files.pythonhosted.org/packages/e5/3f/143b048436775b0f76ac3eec145c019e8173ccc2885c8f20319b996d5e83/charset_normalizer-3.5.1.tar.gz", hash = "sha256:6117b84ea48435e5356dc737f5121485c30920ba43375fa7b434fd753df0eac3", size = 171764, upload-time = "2026-08-15T08:20:44.807Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/71/aa/554e2614f38fc34c58ff1d0911ae8535ad2516440d5482d76fe59f1088b0/charset_normalizer-3.5.1-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:d1ee1e296209fdce05b81b663250eefa02213a2da7b41bf26f7829b8ba3545aa", size = 369072, upload-time = "2026-08-15T08:16:22.964Z" }, - { url = "https://files.pythonhosted.org/packages/03/6d/439231dfc3ccfa6f8c06477b7da2219cbd41a2de3d49084df8ec7b5100f2/charset_normalizer-3.5.1-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:e9fbdce1e47394b09bc9f26ab117dfc8d6491977a11d86f592bb42c779db2fda", size = 251142, upload-time = "2026-08-15T08:16:24.81Z" }, - { url = "https://files.pythonhosted.org/packages/55/53/7d819bd23a00ef45039146fa2cce1daa2f0771e758c5653ee1f6edac91ed/charset_normalizer-3.5.1-cp310-cp310-manylinux2014_armv7l.manylinux_2_17_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:00668ebb0609751758682eb0b5857e7c35b9f00e84dfdef062e103244ec94d45", size = 240714, upload-time = "2026-08-15T08:16:26.392Z" }, - { url = "https://files.pythonhosted.org/packages/b2/2c/45847198c16f4b38090cc7423b2b6a9008e438704d8ab413211832498d31/charset_normalizer-3.5.1-cp310-cp310-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:ba2f37ee79e6338845261a3c5b1784e5d1acdff2c0785b284f1b633033d136ab", size = 279637, upload-time = "2026-08-15T08:16:27.961Z" }, - { url = "https://files.pythonhosted.org/packages/69/2b/d8be3523ddf9f0b0f3e56d1359034aa10653a4d11564c697f802b4775766/charset_normalizer-3.5.1-cp310-cp310-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:ce854f5f478050ade5a238731c4ca985a7d3b3cb53ff600a9b5c3b689b5f0a7a", size = 276543, upload-time = "2026-08-15T08:16:29.399Z" }, - { url = "https://files.pythonhosted.org/packages/32/cd/4f564b8f132de25db594efc706897069f016790cea63a5669c9df2675f64/charset_normalizer-3.5.1-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:96eefc178f8636b9c760c5829345307fd81cfae9ab1e80997dbddeb0f54ee9a3", size = 261644, upload-time = "2026-08-15T08:16:30.722Z" }, - { url = "https://files.pythonhosted.org/packages/f5/e3/38b975422534a608f98c360e79c2f07c763d66dd4272300d45fb1fee54b0/charset_normalizer-3.5.1-cp310-cp310-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:366ec70f5547c640d3ce1985722490f23faf4eb5216a7eeba78277490e78dacb", size = 259609, upload-time = "2026-08-15T08:16:32.248Z" }, - { url = "https://files.pythonhosted.org/packages/87/bd/fbc24d825c66f1c74f6ccdea3742c3d8354a4888e86d1315a197fee69061/charset_normalizer-3.5.1-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:950f23cb393f85543777b0433f082cddd25b51ab398eac7971146495679efe5f", size = 252457, upload-time = "2026-08-15T08:16:33.849Z" }, - { url = "https://files.pythonhosted.org/packages/b9/2d/918d0e98a0e679469ed05bb2d90c2088b4d315bb612969d8499f76fb5210/charset_normalizer-3.5.1-cp310-cp310-musllinux_1_2_armv7l.whl", hash = "sha256:c1dcc36dcb96abc02236e182d17e0f71430152a6c2c7447421da2d2dc144edea", size = 242240, upload-time = "2026-08-15T08:16:35.396Z" }, - { url = "https://files.pythonhosted.org/packages/20/c8/c36f6e0b2dfec351bd38cbc05362697e58bcd073d7dbd95154290c9714ce/charset_normalizer-3.5.1-cp310-cp310-musllinux_1_2_ppc64le.whl", hash = "sha256:07ffd07412fc5d5e84cd8952acf9ff7e4ed7a708e69d1bada19d8ba91711353f", size = 280308, upload-time = "2026-08-15T08:16:36.825Z" }, - { url = "https://files.pythonhosted.org/packages/ca/7b/311b3e02e8c4092400c449c850a760d8c45d900983c83a70cc07208c551d/charset_normalizer-3.5.1-cp310-cp310-musllinux_1_2_riscv64.whl", hash = "sha256:f5542f9b941279d82d41eb0aa9f98eba36fe4df5c7086c651df7944935b37182", size = 258679, upload-time = "2026-08-15T08:16:38.22Z" }, - { url = "https://files.pythonhosted.org/packages/b9/90/082cc45599c392f28c036a497f49e0634041a785fc3849c80ccf396d096f/charset_normalizer-3.5.1-cp310-cp310-musllinux_1_2_s390x.whl", hash = "sha256:a545775cfe815855ea32d7c27731d79da358ef2055b4a25830231b1622dd18aa", size = 277221, upload-time = "2026-08-15T08:16:39.62Z" }, - { url = "https://files.pythonhosted.org/packages/58/ad/b9aecf38d805cbcf84fa94f14c5d972a16561e20296a11dc799a5dcf3763/charset_normalizer-3.5.1-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:494b70049a4d69aec6e8137c13af4cf8db8c9f9820a1392ac293b0dd2987a818", size = 263799, upload-time = "2026-08-15T08:16:40.885Z" }, - { url = "https://files.pythonhosted.org/packages/b7/23/b38a20598d5a825f85d9d7636860e56ff0db1479f86497a6e485aa9326f7/charset_normalizer-3.5.1-cp310-cp310-win32.whl", hash = "sha256:94fbf1c0c6cc0d3d5e50f9a9313a8cdca90dd696d34b381cd1704f8c9e939f20", size = 182037, upload-time = "2026-08-15T08:16:42.198Z" }, - { url = "https://files.pythonhosted.org/packages/d2/21/83fffb77864408b8bf0fe1ca603926401d6f8775a8e150b39aacc9958f8a/charset_normalizer-3.5.1-cp310-cp310-win_amd64.whl", hash = "sha256:be47f99644b208bff7766314013f9acf57b056b04191d570d68ad14022cf5b1d", size = 206030, upload-time = "2026-08-15T08:16:43.787Z" }, - { url = "https://files.pythonhosted.org/packages/86/2e/b93135b5034b1157fb29554b0d06d4844ce62282f0e0a14036f93d7ee2e7/charset_normalizer-3.5.1-cp310-cp310-win_arm64.whl", hash = "sha256:a6d095662e73e74f0a49988e0593373e243e3a52e27bfeea0a859e88acf4a0f5", size = 185092, upload-time = "2026-08-15T08:16:45.177Z" }, { url = "https://files.pythonhosted.org/packages/6a/b6/034f6802e9c3f6418966cfabb7db8c9252cc2429c5098f41cc43af804149/charset_normalizer-3.5.1-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:eda059b6bc8bc0812d626fd91a7ce01bf583df0a61296eff390fd94141a34e30", size = 363585, upload-time = "2026-08-15T08:16:46.646Z" }, { url = "https://files.pythonhosted.org/packages/d5/fa/6a7e2a7c4b5451912b8c417732df79574354443592a88d616de03da66ae5/charset_normalizer-3.5.1-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:aa2bb0b37202dca27175591f761108b5d34096ade1191ffe4808bdf6b1571488", size = 251189, upload-time = "2026-08-15T08:16:48.287Z" }, { url = "https://files.pythonhosted.org/packages/a4/c8/ab42b07cfd82e919f427fcfaa7c41abae8242833ad1aad66d42bae40b669/charset_normalizer-3.5.1-cp311-cp311-manylinux2014_armv7l.manylinux_2_17_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:0b2b1b3fa5670c127b246df1d0c059defd41f689a868a3b9d79df9b1cac42d22", size = 239724, upload-time = "2026-08-15T08:16:49.67Z" }, diff --git a/.github/workflows/python_unit_tests.yml b/.github/workflows/python_unit_tests.yml index a30412bbcd..1482a641c0 100644 --- a/.github/workflows/python_unit_tests.yml +++ b/.github/workflows/python_unit_tests.yml @@ -35,6 +35,9 @@ jobs: unit-test: if: ${{ github.actor != 'dependabot[bot]' }} runs-on: ubuntu-latest + # A full run takes about 3 minutes, so fail a hung test quickly rather than + # holding the runner for the 6 hour default + timeout-minutes: 20 strategy: matrix: python-version: ["3.11", "3.12", "3.13", "3.14"] diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000000..b508f58de9 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,250 @@ +# AGENTS.md + +Guidance for AI coding agents working in this repository. + +## Project Overview + +OpenC3 COSMOS is an open-source command and control system for embedded systems. It provides a web-based UI for telemetry display, command sending, script execution, logging, and more. + +- Docs: https://docs.openc3.com/docs (source in `docs.openc3.com/`, which has its own `CLAUDE.md`) +- Example user projects: [cosmos-project](https://github.com/OpenC3/cosmos-project) (Core), [cosmos-enterprise-project](https://github.com/OpenC3/cosmos-enterprise-project) (Enterprise) + +## Architecture + +### Microservices (Docker Containers) + +- **openc3-cosmos-cmd-tlm-api** - Rails REST API for command/telemetry operations +- **openc3-cosmos-script-runner-api** - Rails API for script execution +- **openc3-operator** - Ruby operator managing interfaces and microservices +- **openc3-buckets** - S3-compatible object storage (Versitygw) for logs and configurations +- **openc3-redis** - Valkey data store and pub/sub messaging +- **openc3-traefik** - Reverse proxy (access at http://localhost:2900) + +Services communicate via Valkey pub/sub and HTTP APIs. WebSockets go through AnyCable. + +### Core Library + +- **openc3/** - Ruby gem with C extensions for performance-critical operations +- **openc3/python/** - Python library (uv-managed) with equivalent functionality + +### Frontend (pnpm Workspace) + +Vue 3 + Vuetify 3, built with Vite. Located in `openc3-cosmos-init/plugins/packages/`: + +- **openc3-tool-base** - Base Vue components for tools +- **openc3-vue-common** - Shared Vue components +- **openc3-js-common** - Shared JavaScript utilities +- **openc3-cosmos-tool-\*** - Individual tools (cmdsender, scriptrunner, tlmviewer, etc.) +- **openc3-cosmos-demo** - Demo plugin with test targets (INST, INST2, EXAMPLE, TEMPLATED) + +### Enterprise vs Core + +Set `ENTERPRISE=1` for Enterprise features. Enterprise uses Keycloak authentication; Core uses simple password authentication. + +## Common Commands + +### Docker + +```bash +./openc3.sh build # Build all containers from source +./openc3.sh start # Build + run (first time setup) +./openc3.sh run # Run containers (after build) +./openc3.sh stop # Stop containers gracefully +./openc3.sh cleanup # WARNING: Deletes all data and volumes +./openc3.sh test # List containerized test suites (rspec, playwright, ...) +./openc3.sh cli help # CLI: generate plugin/target, validate, load, etc. +``` + +### Ruby (openc3 gem) + +```bash +cd openc3 +bundle install +bundle exec rake build # Compile C extensions (required first time) +bundle exec rspec spec/path/to/spec.rb:42 # Run a file or a specific line +``` + +### Rails APIs + +The Rails APIs depend on the local `openc3` gem, which must be built first. Set +`OPENC3_DEVEL` so Bundler resolves it from source instead of RubyGems. + +```bash +export OPENC3_DEVEL=../openc3 +cd openc3-cosmos-cmd-tlm-api && bundle install && bundle exec rspec +cd openc3-cosmos-script-runner-api && bundle install && bundle exec rspec +``` + +### Python + +```bash +cd openc3/python +uv sync # Create venv and install dependencies +uv run pytest test/path/to/test.py # Run a test file +just verify-changed # Format, lint, and test changed files +just verify # Format, lint, and test everything +``` + +Run `just` to list all recipes. Always invoke tools via `uv run` rather than activating the venv. + +### Frontend + +```bash +cd openc3-cosmos-init/plugins +pnpm install +pnpm build:common # Build shared packages first +cd packages/ && pnpm lint # Lint one package (--fix to auto-fix) +``` + +### Playwright E2E + +```bash +cd playwright +pnpm install +./playwright.sh install-playwright +./playwright.sh build-plugin # Required before running tests +pnpm test:parallel # *.p.spec.ts +pnpm test:serial # *.s.spec.ts +pnpm playwright test ./tests/command-sender.p.spec.ts --project=chromium +pnpm test:enterprise # Enterprise suite +PWDEBUG=1 pnpm test:parallel --headed # Debug mode +``` + +## Code Style + +- **Ruby**: RuboCop, configured in `.rubocop.yml`. +- **Python**: Ruff for lint and format, configured in `openc3/python/pyproject.toml`. Minimum supported Python is 3.11 (ruff `target-version = "py311"`), so don't use syntax newer than that. Commit `uv.lock` with any dependency change. +- **JavaScript/Vue**: ESLint 9 + Prettier, configured in `openc3-cosmos-init/plugins/eslint.config.mjs`. After editing Vue/JS files, always run `pnpm lint` in the package. CI runs `pnpm lint --max-warnings 0`, so any warning fails the build. + +## Sharding + +Sharding bugs are easy to miss because nothing fails locally: the default +`.env` and `compose.yaml` have one Redis, one ephemeral Redis, and one TSDB +(no `SHARDNUM` in the hostnames). The test mocks +(`mock_redis` in `spec_helper.rb`, `FakeValkey` in `test_helper.py`) also return +the same instance for every shard. So code that reads shard 0 instead of the +target's shard passes every test and only breaks in multi-shard Enterprise +deployments. Two separate concepts: + +- **`db_shard`** - which Redis / ephemeral Redis / QuestDB instance holds a + target's data. Set per target, interface, router, and microservice with + `DB_SHARD`. Hostnames come from `OPENC3_REDIS_HOSTNAME`, + `OPENC3_REDIS_EPHEMERAL_HOSTNAME`, and `OPENC3_TSDB_HOSTNAME` with `SHARDNUM` + replaced by the shard number. +- **`shard`** - which operator runs a microservice (`SHARD` keyword, + `OPENC3_SHARD` env var, filtered in `microservice_operator.rb`). + +When touching code that reads or writes Redis, topics, or TSDB: + +- **Target data lives on the target's db_shard; everything else is on db_shard 0.** + Packet definitions (`openc3tlm`/`openc3cmd`), CVT, packet counts, and the + `COMMAND`/`DECOMCMD`/`TELEMETRY`/`DECOM` topics are per target. Model configs + like `TargetModel` itself are on 0. Look up the shard with + `Store.db_shard_for_target` or `QuestDBClient.db_shard_for_target`, or use + `TargetModel.store_for_target`. +- **Pass `db_shard:` explicitly.** `Topic`, `Store.instance`, `EphemeralStore.instance`, + and `QuestDBClient` methods default to `db_shard: 0`, so a missing argument + silently uses the wrong instance. +- **Reading many topics:** split them with `Topic.group_topics_by_db_shard` and + read each group from its own instance. A single `read_topics` call can't + span shards. +- **Listing or aggregating:** status and metric models (`InterfaceStatusModel`, + `MicroserviceStatusModel`, `MetricModel`) include `DbShardedModel`, which + iterates all active shards. New per-interface or per-microservice data should + do the same, and queries that merge results across TSDB shards need care. +- **Per-shard state:** `StoreQueued` and `EphemeralStoreQueued` keep one instance + per shard, and some are created lazily after startup. Settings like update + interval, and shutdown or cleanup, must apply to every instance, including + ones created later. +- **Ruby and Python must match.** Most of this logic exists in both + `openc3/lib` and `openc3/python/openc3`; past bugs were fixed in one and not + the other. +- **Tests:** a test that only uses db_shard 0 proves nothing about sharding. + Because the mocks share one instance, assert that the call used the right + `db_shard:` (e.g. `expect(Store).to receive(:instance).with(db_shard: 1)`) + rather than only checking that the data came back. + +## Cross-Repo Changes (Enterprise) + +COSMOS Enterprise (`../cosmos-enterprise`, private) builds on this repo: its +images start from the core images and add the `openc3-enterprise` gem, and its +Python and frontend packages import core's. When changing core, check whether +enterprise depends on what you changed: + +- **Optional enterprise hooks in core.** Core loads enterprise code with + `begin; require 'openc3-enterprise/...'; rescue LoadError` (e.g. `RolesController`, + `utilities/authorization.rb`, `api/api.rb`). Changing one of these files or its + fallback means the enterprise version (`openc3-enterprise-gem/lib/openc3-enterprise/`) + likely needs the same change. Find them with `grep -rn "openc3-enterprise" openc3/lib openc3-cosmos-*-api`. +- **Base classes and modules.** Enterprise subclasses `Model`, `Microservice`, + `Operator`, `Migration`, `Bucket`, `Secrets`, `TargetFile`, and `ModelController`, + and requires many `openc3/models`, `openc3/utilities`, and `openc3/topics` files. + Renaming or changing the signature of anything public there can break enterprise. +- **Python.** `openc3enterprise` imports `openc3.utilities`, `models`, `topics`, + `api`, `io`, and `environment`. +- **Frontend.** Enterprise tools import heavily from `@openc3/vue-common` and + `@openc3/js-common`. Treat their exports as a public API. +- **Environment variables.** A new or renamed variable in `compose.yaml` usually + also belongs in enterprise's `compose.yaml` and `openc3-helm/openc3/templates/_env-vars.tpl`. +- **Playwright.** Enterprise specs live in `cosmos-enterprise/playwright/enterprise` + (the copy under `playwright/tests/enterprise` here is gitignored). UI changes in + shared components can break them. + +To check, grep `../cosmos-enterprise` for the changed class, method, export, or +variable name. If it isn't checked out, say so instead of assuming nothing breaks. +Report the enterprise changes needed; don't edit that repo unless asked. + +## Ruby and Python Parity + +Most of the core library exists in both languages, and users can write targets, +interfaces, and scripts in either. When changing one, search for the equivalent +in the other and change it too, including its tests: + +| Ruby | Python | +| ---- | ------ | +| `openc3/lib/openc3//.rb` | `openc3/python/openc3//.py` | +| `openc3/spec//_spec.rb` | `openc3/python/test//test_.py` | + +- Class and method names usually match (`CamelCase` classes, `snake_case` + methods), so grep the other tree for the name. Ruby keyword arguments + (`scope:`, `db_shard:`) are Python keyword arguments (`scope=`, `db_shard=`). +- Code under `openc3/ext/` is a Ruby C extension; its Python equivalent is plain + Python (e.g. `ext/openc3/ext/structure` → `python/openc3/packets/structure.py`). +- Some Python modules have a `.pyi` stub next to them (`store_queued`, + `store_implementation`, `logger`). Update the stub when the signature changes. +- Some Ruby areas have no Python equivalent (`operators/`, `migrations/`, + `core_ext/`, `win32/`); a missing match there is expected. If you can't find + an equivalent elsewhere, say so instead of skipping it silently. + +## Scripting API + +The scripting API is the public interface users call from scripts, tools, and +microservices. It is defined in: + +- `openc3/lib/openc3/api/*_api.rb` (methods registered in `WHITELIST`) and + `openc3/lib/openc3/script/*.rb` +- `openc3/python/openc3/api/` and `openc3/python/openc3/script/` (keep in sync + per Ruby and Python Parity above) + +When you add, remove, rename, or change the parameters, return value, or +behavior of one of these methods, update +`docs.openc3.com/docs/guides/scripting-api.md` in the same change. Each method +there has its own `###` heading with: + +- a `Since X.Y.Z` badge. For a new method, write the version as the placeholder + `_Coming Soon_`, never a version number: + `Since _Coming Soon_`. + `scripts/release/openc3_set_versions.rb` replaces it with the release version + on a production release. (The same placeholder works for `since:` in the + `openc3/data/config/*.yaml` files.) +- Python and Ruby syntax tabs, a parameter table, and Python and Ruby example tabs + +Removed or deprecated methods go in the migration table near the top of the +file. See `docs.openc3.com/CLAUDE.md` for the docs build and +`check_doc_code_examples.py` for checking the examples. + +## Commit Messages + +[Conventional Commits v1.0.0](https://www.conventionalcommits.org/en/v1.0.0/). +`.claude/commands/commit-message.md` is the authority on format (Claude Code +users can run `/commit-message`); follow it rather than duplicating rules here. diff --git a/CLAUDE.md b/CLAUDE.md index 45bc18bf8c..43c994c2d3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,287 +1 @@ -# CLAUDE.md - -This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. - -## Project Overview - -OpenC3 COSMOS is an open-source command and control system for embedded systems. It provides a web-based UI for telemetry display, command sending, script execution, logging, and more. The system is designed for testing, operating, and monitoring embedded systems that communicate via TCP/IP, UDP, Serial, etc. - -## Documentation - -The COSMOS documentation can be found at https://docs.openc3.com/docs. To get started with COSMOS Core reference https://github.com/OpenC3/cosmos-project and to get started with COSMOS Enterprise reference https://github.com/OpenC3/cosmos-enterprise-project. - -## Architecture - -### Microservices (Docker Containers) - -- **openc3-cosmos-cmd-tlm-api** - Rails 7.2 REST API for command/telemetry operations -- **openc3-cosmos-script-runner-api** - Rails 7.2 API for script execution -- **openc3-operator** - Ruby operator managing interfaces and microservices -- **openc3-buckets** - S3-compatible object storage for logs and configurations -- **openc3-redis** - Valkey data store and pub/sub messaging -- **openc3-traefik** - Reverse proxy (access at http://localhost:2900) - -### Core Library - -- **openc3/** - Ruby gem with ~40 C extensions for performance-critical operations -- **openc3/python/** - Python library (uv-managed) with equivalent functionality - -### Frontend (pnpm Workspace) - -Located in `openc3-cosmos-init/plugins/packages/`: - -- **openc3-tool-base** - Base Vue 3 components for tools -- **openc3-vue-common** - Shared Vue 3 components (Vuetify 3) -- **openc3-js-common** - Shared JavaScript utilities -- **openc3-cosmos-tool-\*** - Individual tool packages (cmdsender, scriptrunner, tlmviewer, etc.) -- **openc3-cosmos-demo** - Demo plugin with test targets (INST, INST2, EXAMPLE, TEMPLATED) - -### Communication Flow - -Services communicate via Valkey pub/sub and HTTP APIs. WebSocket support via AnyCable. - -## Common Commands - -### Docker Management (Primary Development Method) - -```bash -./openc3.sh build # Build all containers from source -./openc3.sh start # Build + run (first time setup) -./openc3.sh run # Run containers (after build) -./openc3.sh stop # Stop containers gracefully -./openc3.sh cleanup # WARNING: Deletes all data and volumes -``` - -### CLI Commands - -```bash -./openc3.sh cli help # Show CLI help -./openc3.sh cli generate plugin MyPlugin # Generate new plugin -./openc3.sh cli generate target MY_TARGET # Generate new target -./openc3.sh cli validate myplugin.gem # Validate plugin -./openc3.sh cli load myplugin.gem # Load plugin -``` - -### Ruby Tests - -```bash -cd openc3 -bundle install -bundle exec rake build # Compile C extensions (required first time) -bundle exec rspec # Run all tests -bundle exec rspec spec/path/to/spec.rb # Run single test file -bundle exec rspec spec/path/to/spec.rb:42 # Run specific line -``` - -### Python Development - -#### First Time Setup - -Install required tools (macOS): - -```bash -brew install uv ruff just -``` - -Setup Python environment: - -```bash -cd openc3/python -uv python install # Install Python from .python-version -uv sync # Create venv and install dependencies -``` - -#### Common Python Commands - -**Using Just (recommended):** - -```bash -just # Show all available commands -just test # Run all tests -just test-cov # Run tests with coverage -just lint # Check code quality -just format # Format all code -just verify # Format, lint, and test everything -just verify-changed # Fast check of only changed files -just add PACKAGE # Add new dependency -just update # Update all dependencies -``` - -**Manual commands:** - -```bash -uv run pytest # Run all tests -uv run pytest test/path/to/test.py # Run single test file -uv run ruff check openc3 # Lint -uv run ruff format openc3 # Format -uv run ruff check openc3 --fix # Lint and auto-fix -uv run coverage run -m pytest && uv run coverage report # Coverage -``` - -#### UV Workflow Tips - -**Adding dependencies:** - -```bash -uv add package-name # Add runtime dependency -uv add --dev package-name # Add dev dependency -uv sync # Install after manual pyproject.toml edits -``` - -**Updating dependencies:** - -```bash -uv lock --upgrade # Update uv.lock -uv sync # Install updated dependencies -``` - -**Managing Python versions:** - -```bash -uv python list # List available Python versions -uv python install 3.13 # Install specific Python version -uv python pin 3.13 # Update .python-version -``` - -**Troubleshooting:** - -```bash -rm -rf .venv && uv sync # Reset virtual environment -uv cache clean # Clear uv cache -uv python list --only-installed # Show installed Python versions -``` - -**Migration from Poetry:** -See `openc3/python/README-DEV.md` for detailed migration instructions. - -### Frontend (Vue.js/Vuetify) - -```bash -cd openc3-cosmos-init/plugins -pnpm install -pnpm build:common # Build shared packages first -pnpm lint # ESLint, run from the openc3-cosmos-init/plugins/packages/*tool*/ directory -``` - -### API Tests (Rails) - -The Rails APIs depend on the local `openc3` gem, which must be built first (see -Ruby Tests above). Set `OPENC3_DEVEL` to the path of the local `openc3` gem so -Bundler resolves it from source instead of trying to fetch it from RubyGems. - -```bash -export OPENC3_DEVEL=../openc3 -cd openc3-cosmos-cmd-tlm-api && bundle install && bundle exec rspec -cd openc3-cosmos-script-runner-api && bundle install && bundle exec rspec -``` - -### Playwright E2E Tests - -```bash -cd playwright -pnpm install -./playwright.sh install-playwright -./playwright.sh build-plugin # Required before running tests -pnpm test # Run all tests -pnpm test:parallel # Run parallel tests (*.p.spec.ts) -pnpm test:serial --workers=1 # Run serial tests (*.s.spec.ts) -pnpm playwright test ./tests/command-sender.p.spec.ts --project=chromium # Single file -ENTERPRISE=1 pnpm test:parallel # Enterprise edition tests -pnpm test:parallel --headed # Visible browser -PWDEBUG=1 pnpm test:parallel --headed # Debug mode -``` - -### Test Commands via Docker - -```bash -./openc3.sh test rspec # Run Ruby tests in container -./openc3.sh test playwright # Run Playwright tests -``` - -## Technology Stack - -- **Ruby 3.4** - Backend APIs and core library -- **Rails 7.2** - REST APIs with AnyCable for WebSockets -- **Python 3.10-3.14** - Alternative scripting language -- **UV** - Python package manager (fast Rust-based replacement for pip/poetry) -- **Ruff** - Python linter and formatter (fast Rust-based replacement for flake8/black) -- **Vue.js 3 + Vuetify 3** - Frontend UI framework -- **Vite** - Frontend build tool -- **pnpm 11** - Frontend package management (monorepo workspace) -- **Node.js 24** - JavaScript runtime -- **Docker Compose** - Container orchestration -- **Valkey** - Caching, pub/sub, ephemeral state -- **Versitygw** - S3-compatible object storage -- **Playwright** - E2E testing - -## Code Style - -### Ruby - -- RuboCop configured in `.rubocop.yml` (many rules disabled) -- Target Ruby version: 3.4 - -### Python - -- **Package Manager**: UV (fast Rust-based replacement for pip/poetry) -- **Linter & Formatter**: Ruff (replaces flake8, black, isort) -- **Active Rules**: E, F, I, N, UP, B, W, C4, SIM (pycodestyle, Pyflakes, isort, naming, pyupgrade, bugbear, simplify) -- **Line Length**: 120 -- **Target Python**: 3.10 (minimum ruff target for backwards compatibility) -- **Docker Python**: 3.12 (used in Alpine/UBI base images) -- **Supported Versions**: 3.11, 3.12, 3.13 (Docker version ±1) -- **Config**: `openc3/python/pyproject.toml` -- **Lockfile**: `openc3/python/uv.lock` (replaces poetry.lock) -- **Python Version**: `openc3/python/.python-version` (default: 3.12) - -**Important UV Notes:** - -- Always use `uv sync --frozen` in CI/CD and Dockerfiles (fails if lockfile is out of sync) -- Use `uv sync --locked` in development (updates if out of sync) -- Use `uv run` instead of activating the venv manually -- The lockfile (`uv.lock`) must be committed to git -- Don't commit `.venv` directory (it's in .gitignore) -- Python microservices use `OPENC3_PYTHON_BIN` environment variable (defaults to `/openc3/python/.venv/bin/python`) to support UV's editable installs - -### JavaScript/TypeScript - -- ESLint 9 with Vue parser -- Prettier for formatting (enforced via `prettier/prettier` ESLint rule) -- Config in `openc3-cosmos-init/plugins/eslint.config.mjs` -- **After editing Vue/JS files, always run the linter** to catch formatting issues before committing: - ```bash - cd openc3-cosmos-init/plugins/packages/ - pnpm lint # check for issues - pnpm lint --fix # auto-fix formatting - ``` -- CI runs `pnpm lint --max-warnings 0` — any warning fails the build - -## Plugin Development - -Plugins extend COSMOS with new targets, tools, and interfaces. Located in `openc3-cosmos-init/plugins/packages/`. - -Each plugin is a Ruby gem containing: - -- Target definitions (commands, telemetry, screens) -- Optional Vue.js tools -- Interface implementations - -Generate new plugins: `./openc3.sh cli generate plugin MyPlugin` - -## Enterprise vs Core - -Set `ENTERPRISE=1` environment variable for Enterprise features. Enterprise uses Keycloak authentication. Core uses simple password authentication. - -## Commit Message Format - -[Conventional Commits v1.0.0](https://www.conventionalcommits.org/en/v1.0.0/). -Run `/commit-message` to generate one from the staged changes - -`.claude/commands/commit-message.md` is the authority on format, so the rules -live in one place rather than being duplicated here. - -**Guidelines:** - -- Subject line: imperative mood, lowercase, no period, max 72 chars -- Body: explain what and why (not how), keep concise (2-4 lines max) -- Scope: component/module affected -- Always include Claude Code attribution footer +@AGENTS.md diff --git a/docs.openc3.com/docs/guides/bridges.md b/docs.openc3.com/docs/guides/bridges.md index 00c2edbb5d..68486fed15 100644 --- a/docs.openc3.com/docs/guides/bridges.md +++ b/docs.openc3.com/docs/guides/bridges.md @@ -26,7 +26,7 @@ protocol like LENGTH. -- Requires a host Python 3.10+ installation +- Requires a host Python 3.11+ installation - Install the OpenC3 package (recommend using a venv) ```bash diff --git a/examples/throughput_server/README.md b/examples/throughput_server/README.md index 8dfeab275c..0ff111cfa3 100644 --- a/examples/throughput_server/README.md +++ b/examples/throughput_server/README.md @@ -30,7 +30,7 @@ COSMOS Container Throughput Server (Python) ## Requirements -- Python 3.10 or higher +- Python 3.11 or higher - No external dependencies (uses only standard library) ## Usage @@ -50,44 +50,44 @@ python throughput_server.py --debug ### Command Line Options -| Option | Default | Description | -|--------|---------|-------------| -| `--inst-port` | 7778 | Port for INST (Ruby) target | -| `--inst2-port` | 7780 | Port for INST2 (Python) target | -| `--debug` | False | Enable debug logging | +| Option | Default | Description | +| -------------- | ------- | ------------------------------ | +| `--inst-port` | 7778 | Port for INST (Ruby) target | +| `--inst2-port` | 7780 | Port for INST2 (Python) target | +| `--debug` | False | Enable debug logging | ## Supported Commands The server responds to the following CCSDS commands: -| Command | PKTID | Description | -|---------|-------|-------------| -| START_STREAM | 200 | Start telemetry streaming at specified rate | -| STOP_STREAM | 201 | Stop telemetry streaming | -| GET_STATS | 202 | Request THROUGHPUT_STATUS telemetry packet | -| RESET_STATS | 203 | Reset all throughput statistics | +| Command | PKTID | Description | +| ------------ | ----- | ------------------------------------------- | +| START_STREAM | 200 | Start telemetry streaming at specified rate | +| STOP_STREAM | 201 | Stop telemetry streaming | +| GET_STATS | 202 | Request THROUGHPUT_STATUS telemetry packet | +| RESET_STATS | 203 | Reset all throughput statistics | ### START_STREAM Payload -| Field | Type | Description | -|-------|------|-------------| -| RATE | 32-bit UINT | Packets per second (1-100000) | +| Field | Type | Description | +| ------------ | ----------- | --------------------------------------- | +| RATE | 32-bit UINT | Packets per second (1-100000) | | PACKET_TYPES | 32-bit UINT | Bitmask of packet types (default: 0x01) | ## Telemetry Packets ### THROUGHPUT_STATUS (APID 100) -| Field | Type | Description | -|-------|------|-------------| -| CMD_RECV_COUNT | 32-bit UINT | Total commands received | -| CMD_RECV_RATE | 32-bit FLOAT | Commands per second | -| TLM_SENT_COUNT | 32-bit UINT | Total telemetry packets sent | -| TLM_SENT_RATE | 32-bit FLOAT | Telemetry packets per second | -| TLM_TARGET_RATE | 32-bit UINT | Configured streaming rate | -| BYTES_RECV | 64-bit UINT | Total bytes received | -| BYTES_SENT | 64-bit UINT | Total bytes sent | -| UPTIME_SEC | 32-bit UINT | Server uptime in seconds | +| Field | Type | Description | +| --------------- | ------------ | ---------------------------- | +| CMD_RECV_COUNT | 32-bit UINT | Total commands received | +| CMD_RECV_RATE | 32-bit FLOAT | Commands per second | +| TLM_SENT_COUNT | 32-bit UINT | Total telemetry packets sent | +| TLM_SENT_RATE | 32-bit FLOAT | Telemetry packets per second | +| TLM_TARGET_RATE | 32-bit UINT | Configured streaming rate | +| BYTES_RECV | 64-bit UINT | Total bytes received | +| BYTES_SENT | 64-bit UINT | Total bytes sent | +| UPTIME_SEC | 32-bit UINT | Server uptime in seconds | ## CCSDS Packet Format @@ -133,16 +133,17 @@ To use the throughput server with COSMOS, install the modified DEMO plugin with ### Plugin Variables -| Variable | Default | Description | -|----------|---------|-------------| -| `use_throughput_server` | false | Enable throughput server mode | +| Variable | Default | Description | +| ------------------------ | -------------------- | ----------------------------- | +| `use_throughput_server` | false | Enable throughput server mode | | `throughput_server_host` | host.docker.internal | Hostname of throughput server | -| `inst_throughput_port` | 7778 | Port for INST connection | -| `inst2_throughput_port` | 7780 | Port for INST2 connection | +| `inst_throughput_port` | 7778 | Port for INST connection | +| `inst2_throughput_port` | 7780 | Port for INST2 connection | ### Running Throughput Tests 1. Start the throughput server: + ```bash python throughput_server.py ``` diff --git a/examples/throughput_server/throughput_server.py b/examples/throughput_server/throughput_server.py index f706dc8837..57eba658b6 100644 --- a/examples/throughput_server/throughput_server.py +++ b/examples/throughput_server/throughput_server.py @@ -104,7 +104,7 @@ async def handle(self) -> None: self._buffer += data await self._process_buffer() - except asyncio.TimeoutError: + except TimeoutError: # No data, continue loop (allows checking self.running) continue except ConnectionResetError: diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_channel.rb new file mode 100644 index 0000000000..c4039e9b90 --- /dev/null +++ b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_channel.rb @@ -0,0 +1,82 @@ +# encoding: ascii-8bit + +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +module ApplicationCable + # Shared lifecycle for channels that stream from a per-subscription + # broadcaster object. Including channels declare a key prefix with + # `broadcaster_prefix` and implement a private `create_broadcaster`. + # Broadcasters are tracked in BroadcasterRegistry so a subscription whose + # Disconnect never arrives is eventually reaped. + module BroadcasterChannel + extend ActiveSupport::Concern + + # Sent to a heartbeating client whose broadcaster no longer exists (reaped + # or lost), telling it to resubscribe. Handled inside the frontend Cable + # wrapper so it never reaches the channel's received callback. + RESUBSCRIBE_MESSAGE = { '__openc3_cable__' => 'resubscribe' }.freeze + + included do + # Identifies the broadcaster this subscription registered. Kept in + # AnyCable channel state because channel instances don't survive between + # the subscribe and unsubscribe RPCs. + state_attr_accessor :broadcaster_token + end + + class_methods do + def broadcaster_prefix(prefix = nil) + @broadcaster_prefix = prefix if prefix + @broadcaster_prefix + end + end + + def subscribed + # Defensive: if the auth before_subscribe callback rejected us, skip work. + return if subscription_rejected? + stream_from subscription_key + self.broadcaster_token = SecureRandom.uuid + # Clients that heartbeat say so when they subscribe so they can be reaped + # even if they die before sending their first heartbeat + BroadcasterRegistry.register(subscription_key, create_broadcaster(), + heartbeat: [true, 'true'].include?(params[:heartbeat]), token: broadcaster_token) + end + + def unsubscribed + # No token means subscribed was rejected and nothing was registered + token = broadcaster_token + return unless token + stop_stream_from subscription_key + # Only remove our own registration. A resubscribe on the same connection + # shares the key, and if its subscribe is handled before this unsubscribe + # the newer broadcaster must survive. + BroadcasterRegistry.unregister(subscription_key, token: token) + end + + def heartbeat(_data = nil) + unless BroadcasterRegistry.touch(subscription_key, heartbeat: true) + transmit(RESUBSCRIBE_MESSAGE) + end + end + + private + + def subscription_key + "#{self.class.broadcaster_prefix}_#{uuid}" + end + + # Any perform from the client also counts as activity + def broadcaster + BroadcasterRegistry.touch(subscription_key) + BroadcasterRegistry.get(subscription_key) + end + end +end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_registry.rb b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_registry.rb new file mode 100644 index 0000000000..dc8640c422 --- /dev/null +++ b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_registry.rb @@ -0,0 +1,182 @@ +# encoding: ascii-8bit + +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +module ApplicationCable + # Process-wide registry of the per-subscription broadcaster objects + # (StreamingApi, MessagesApi, *EventsApi) that the channels create. + # + # Entries are normally removed by the channel's `unsubscribed`. Under AnyCable + # that arrives as a separate Disconnect RPC from anycable-go, and if it is + # never delivered (anycable-go restarting or crashing, the gRPC stream being + # torn down) the broadcaster and its Redis-reading threads would live for the + # rest of the process. To bound that, clients send a periodic `heartbeat` + # perform and a background sweeper reaps any entry that has stopped + # heartbeating for longer than TTL. + # + # Only entries registered as heartbeating (the client said it heartbeats when + # it subscribed) or that have sent a heartbeat are reaped. Older clients (and + # the Ruby/Python scripting WebSocket APIs) never heartbeat, and a realtime + # stream legitimately goes a long time with no client-to-server traffic, so + # reaping them on idle time alone would kill healthy streams. + class BroadcasterRegistry + # Clients heartbeat every 60s, so this tolerates several missed heartbeats + # (e.g. background tab timer throttling) before reaping + DEFAULT_TTL = 300 + DEFAULT_SWEEP_INTERVAL = 60 + # A ttl shorter than the client heartbeat interval would reap healthy + # clients between heartbeats, so never go below two heartbeats + MIN_TTL = 120 + # Keeps a zero or invalid interval from turning the sweeper into a busy loop + MIN_SWEEP_INTERVAL = 1 + + Entry = Struct.new(:broadcaster, :last_activity, :heartbeat, :token) + + @entries = {} + @mutex = Mutex.new + @sweeper = nil + @clamp_warnings = {} + + class << self + def ttl + env_seconds('OPENC3_CABLE_SUBSCRIPTION_TTL', DEFAULT_TTL, MIN_TTL) + end + + def sweep_interval + env_seconds('OPENC3_CABLE_SWEEP_INTERVAL', DEFAULT_SWEEP_INTERVAL, MIN_SWEEP_INTERVAL) + end + + # heartbeat: true makes the entry reapable immediately rather than only + # after its first heartbeat, so a client that dies right after + # subscribing is still reaped. token identifies this registration so a + # late unregister from an older subscription can't remove a newer one. + def register(key, broadcaster, heartbeat: false, token: nil) + previous = nil + @mutex.synchronize do + previous = @entries[key]&.broadcaster + @entries[key] = Entry.new(broadcaster, now(), heartbeat, token) + end + # A duplicate subscribe for the same key must not leak the old one + kill(key, previous) if previous and !previous.equal?(broadcaster) + start_sweeper() + broadcaster + end + + def get(key) + @mutex.synchronize { @entries[key]&.broadcaster } + end + + def registered?(key) + @mutex.synchronize { @entries.key?(key) } + end + + # Record client activity. Returns false if the key is not registered, + # e.g. because it was already reaped. + def touch(key, heartbeat: false) + @mutex.synchronize do + entry = @entries[key] + return false unless entry + entry.last_activity = now() + entry.heartbeat = true if heartbeat + return true + end + end + + # When token is given only the registration made with that token is + # removed. Returns whether an entry was removed. + def unregister(key, token: nil) + entry = @mutex.synchronize do + current = @entries[key] + next nil unless current + next nil if token and current.token != token + @entries.delete(key) + end + kill(key, entry.broadcaster) if entry and entry.broadcaster + !entry.nil? + end + + # Remove and kill every heartbeating entry idle for longer than ttl. + # Returns the reaped keys. + def sweep(current_time = now()) + expired = [] + @mutex.synchronize do + cutoff = current_time - ttl() + @entries.each do |key, entry| + expired << [key, entry] if entry.heartbeat and entry.last_activity < cutoff + end + expired.each { |key, _| @entries.delete(key) } + end + # Kill outside the lock as it may wait on the broadcaster's threads + expired.each do |key, entry| + OpenC3::Logger.warn("Reaping cable subscription #{key}: no heartbeat in #{(current_time - entry.last_activity).round}s") + kill(key, entry.broadcaster) if entry.broadcaster + end + expired.map(&:first) + end + + def size + @mutex.synchronize { @entries.length } + end + + # For tests + def clear + @mutex.synchronize do + @entries.clear + @clamp_warnings.clear + end + end + + private + + def now + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + def env_seconds(name, default, minimum) + raw = ENV.fetch(name, nil) + return default.to_f if raw.nil? or raw.to_s.strip.empty? + value = Float(raw, exception: false) + return value if value and value >= minimum + fallback = value ? minimum.to_f : default.to_f + # Only warn once per value, ttl is read on every sweep + unless @clamp_warnings[name] == raw + @clamp_warnings[name] = raw + OpenC3::Logger.warn("#{name}=#{raw} is invalid or below the minimum of #{minimum}s, using #{fallback}s") + end + fallback + end + + def kill(key, broadcaster) + broadcaster.kill + rescue => e + OpenC3::Logger.error("Error killing cable subscription #{key}: #{e.formatted}") + end + + def start_sweeper + return if @sweeper&.alive? + @mutex.synchronize do + return if @sweeper&.alive? + @sweeper = Thread.new do + loop do + sleep(sweep_interval()) + begin + sweep() + rescue => e + OpenC3::Logger.error("Cable subscription sweeper error: #{e.formatted}") + end + end + end + end + end + end + end +end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/channel.rb index 00bb90f210..a346c3b08d 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/channel.rb @@ -19,6 +19,14 @@ class Channel < ActionCable::Channel::Base before_subscribe :authenticate_subscription! + # The frontend Cable sends a periodic heartbeat on every subscription. + # Accept it on every channel so ones that don't track liveness (see + # BroadcasterChannel) don't log it as an unknown action. + def heartbeat(_data = nil) + # Intentionally a no-op: channels without a broadcaster have no + # liveness to record. BroadcasterChannel overrides this. + end + private # Authenticate the subscription using either: diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/autonomic_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/autonomic_events_channel.rb index 9ec42dfdf1..a4b8899b2c 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/autonomic_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/autonomic_events_channel.rb @@ -16,23 +16,12 @@ # if purchased from OpenC3, Inc. class AutonomicEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'autonomic_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "autonomic_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = AutonomicEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "autonomic_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + AutonomicEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/calendar_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/calendar_events_channel.rb index 3000a8efd9..f6c23f6813 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/calendar_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/calendar_events_channel.rb @@ -16,23 +16,12 @@ # if purchased from OpenC3, Inc. class CalendarEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'calendar_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "calendar_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = CalendarEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "calendar_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + CalendarEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/config_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/config_events_channel.rb index 7e9b3ee399..4969b1c9db 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/config_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/config_events_channel.rb @@ -12,23 +12,12 @@ # if purchased from OpenC3, Inc. class ConfigEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'config_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "config_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = ConfigEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "config_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + ConfigEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/limits_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/limits_events_channel.rb index af9fce4651..7591fe4faf 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/limits_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/limits_events_channel.rb @@ -16,23 +16,12 @@ # if purchased from OpenC3, Inc. class LimitsEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'limits_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "limits_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = LimitsEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "limits_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + LimitsEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/messages_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/messages_channel.rb index 8578821f63..1c39be7339 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/messages_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/messages_channel.rb @@ -12,15 +12,13 @@ # if purchased from OpenC3, Inc. class MessagesChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'messages' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "messages_#{uuid}" - stream_from subscription_key + private - @@broadcasters[subscription_key] = MessagesApi.new( + def create_broadcaster + MessagesApi.new( subscription_key, params["history_count"], start_offset: params["start_offset"], @@ -31,14 +29,4 @@ def subscribed scope: scope ) end - - def unsubscribed - subscription_key = "messages_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end - end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/notebook_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/notebook_events_channel.rb index 17e13a0d40..b37db5471d 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/notebook_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/notebook_events_channel.rb @@ -7,23 +7,12 @@ # if purchased from OpenC3, Inc. class NotebookEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'notebook_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "notebook_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = NotebookEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "notebook_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + NotebookEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/queue_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/queue_events_channel.rb index 48ee9700b0..eaa01b9b9a 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/queue_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/queue_events_channel.rb @@ -12,23 +12,12 @@ # if purchased from OpenC3, Inc. class QueueEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'queue_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "queue_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = QueueEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "queue_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + QueueEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/streaming_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/streaming_channel.rb index 755c9c1e70..7e2e4969b5 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/streaming_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/streaming_channel.rb @@ -16,25 +16,8 @@ # if purchased from OpenC3, Inc. class StreamingChannel < ApplicationCable::Channel - @@broadcasters = {} - - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "streaming_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = StreamingApi.new(subscription_key, scope: scope) - end - - def unsubscribed - subscription_key = "streaming_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end - end + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'streaming' # data holds the following keys: # start_time - nsec_since_epoch - null for realtime @@ -42,7 +25,6 @@ def unsubscribed # items [Array of Item keys] ie ["DECOM__TLM__INST__ADCS__Q1__RAW"] # scope def add(data) - subscription_key = "streaming_#{uuid}" if validate_data(data) begin # Nil-guard the broadcaster: it only exists between `subscribed` (which @@ -51,7 +33,7 @@ def add(data) # raise NoMethodError here, get rescued below, and reject_subscription — # killing every panel on the connection. A no-op is the correct response; # the client re-subscribes and replays its adds. - @@broadcasters[subscription_key]&.add(data) + broadcaster()&.add(data) rescue OpenC3::AuthError, OpenC3::ForbiddenError transmit({ "error" => "unauthorized" }) reject() # Sets the rejected state on the connection @@ -68,13 +50,12 @@ def add(data) # items [Array of Item keys] ie ["DECOM__TLM__INST__ADCS__Q1__RAW"] # scope def remove(data) - subscription_key = "streaming_#{uuid}" if validate_data(data) begin # Nil-guard the broadcaster (see `add`): a `remove` that races a # rejected/torn-down subscription must be a harmless no-op, not a # NoMethodError that reject_subscription()s every panel on the connection. - @@broadcasters[subscription_key]&.remove(data) + broadcaster()&.remove(data) rescue OpenC3::AuthError, OpenC3::ForbiddenError transmit({ "error" => "unauthorized" }) reject() # Sets the rejected state on the connection @@ -89,6 +70,10 @@ def remove(data) private + def create_broadcaster + StreamingApi.new(subscription_key, scope: scope) + end + def validate_data(data) result = true unless data['scope'] diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/system_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/system_events_channel.rb index de699cb39e..70add0d41e 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/system_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/system_events_channel.rb @@ -12,23 +12,12 @@ # if purchased from OpenC3, Inc. class SystemEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'system_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "system_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = SystemEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "system_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + SystemEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/timeline_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/timeline_events_channel.rb index 6900b1c00c..4dfc04d39a 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/timeline_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/timeline_events_channel.rb @@ -16,23 +16,12 @@ # if purchased from OpenC3, Inc. class TimelineEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'timeline_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "timeline_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = TimelineEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "timeline_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + TimelineEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb index efdbf9bcb6..c48e1cf607 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb @@ -18,11 +18,148 @@ require_relative 'streaming_thread' class RealtimeStreamingThread < StreamingThread + # How far behind realtime (in seconds) the stream may fall before we consider + # skipping ahead. Set OPENC3_STREAMING_MAX_LAG_SECONDS to 0 to disable. + DEFAULT_MAX_LAG_SECONDS = 10.0 + # Beyond this multiple of the max lag we skip immediately, even if catching up, + # because Redis only retains a limited window and the data is about to be trimmed + SKIP_IMMEDIATELY_FACTOR = 6 + # Minimum seconds between lag checks. Each check costs a Redis round trip per + # db_shard so don't do it every pass; lag thresholds are many seconds anyway. + LAG_CHECK_INTERVAL_SECONDS = 1.0 + def initialize(streaming_api, collection, scope: nil) super(streaming_api, collection, scope: scope) + @max_lag_seconds = max_lag_seconds() + # Monotonic time when the current "behind" grace window started (nil when caught up) + @behind_since = nil + # Lag measured at the start of the current grace window, compared against + # at the end of the window to decide whether we're catching up + @behind_lag = nil + # Monotonic time of the last lag check, used to rate limit the Redis queries + @last_lag_check = nil end def thread_body redis_thread_body() end + + # Realtime displays want current values, so if we can't keep up we jump to the + # newest data rather than falling ever further behind. A stream that is over + # the threshold but catching up (e.g. just handed off from a historical query) + # gets a grace window of @max_lag_seconds, and each window in which the lag + # shrinks starts a new one. Only when the lag fails to shrink over a whole + # window (or gets extreme) do we skip. + # + # Lag is how far the newest msg_id we read trails the newest msg_id in the + # stream. Both are stamped by the same Redis server (XADD '*') so the result + # is immune to clock skew between this container and Redis. + def check_lag(newest_msg_ids, item_objects_by_topic, packet_objects_by_topic) + # Disabled, or nothing was read this pass so there is nothing to measure. + # An idle stream can't be behind: the blocking read would have returned data. + return if @max_lag_seconds <= 0 or newest_msg_ids.empty? + + # Monotonic clock so wall clock adjustments (NTP, DST) can't affect the windows + now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + return if @last_lag_check and (now - @last_lag_check) < LAG_CHECK_INTERVAL_SECONDS + @last_lag_check = now + + heads = stream_heads(newest_msg_ids.keys, item_objects_by_topic, packet_objects_by_topic) + topic_lags = {} + newest_msg_ids.each do |topic, msg_id| + head = heads[topic] + # Stream was deleted or trimmed to empty since we read it - can't measure + next unless head + # msg_ids are "-" so the difference of the ms parts is the lag + topic_lags[topic] = (msg_id_ms(head) - msg_id_ms(msg_id)) / 1000.0 + end + # The stream as a whole is only as current as its slowest topic + lag = topic_lags.values.max || 0.0 + if lag <= @max_lag_seconds + # Within tolerance so reset any grace window in progress + @behind_since = nil + @behind_lag = nil + return + end + + # Moderately behind: apply the grace window logic. Extremely behind falls + # through to skip immediately since waiting risks reading trimmed data. + if lag < @max_lag_seconds * SKIP_IMMEDIATELY_FACTOR + if @behind_since.nil? + # Just fell behind - start the first grace window + @behind_since = now + @behind_lag = lag + return + end + # Still inside the current grace window - keep reading normally + return if (now - @behind_since) < @max_lag_seconds + if lag < @behind_lag + # Catching up - give it another window + @behind_since = now + @behind_lag = lag + return + end + # A whole window passed without the lag shrinking - fall through and skip + end + + skip_ahead(lag, topic_lags, heads, item_objects_by_topic, packet_objects_by_topic) + # Skipping puts us at the head so start fresh on the next check + @behind_since = nil + @behind_lag = nil + end + + # Only topics that are actually behind are advanced so up to date topics + # don't lose messages that arrived during this pass + def skip_ahead(lag, topic_lags, heads, item_objects_by_topic, packet_objects_by_topic) + OpenC3::Logger.warn("Realtime streaming fell #{lag.round(1)}s behind, skipping ahead to the latest data") + topic_lags.each do |topic, topic_lag| + next if topic_lag <= @max_lag_seconds + # Both item and packet subscriptions can share a topic so move them all + objects = (item_objects_by_topic[topic] || []) + (packet_objects_by_topic[topic] || []) + # The next XREAD returns entries after the offset, so setting it to the + # head msg_id resumes with the first message published after this check + objects.each { |object| object.offset = heads[topic] } + end + # Tell opted in clients so they can show the gap rather than silently dropping data + @streaming_api.transmit_notice('SKIPPED', "Streaming fell behind realtime, skipped #{lag.round(1)}s of data", skipped_seconds: lag.round(3)) + end + + # Only an explicit 0 disables lag handling. A malformed value (which to_f + # would silently turn into 0) falls back to the default instead. + def max_lag_seconds + raw = ENV.fetch('OPENC3_STREAMING_MAX_LAG_SECONDS', nil) + return DEFAULT_MAX_LAG_SECONDS if raw.nil? or raw.strip.empty? + value = Float(raw, exception: false) + return value if value + OpenC3::Logger.warn("Invalid OPENC3_STREAMING_MAX_LAG_SECONDS '#{raw}', using #{DEFAULT_MAX_LAG_SECONDS}") + DEFAULT_MAX_LAG_SECONDS + end + + # Returns the newest msg_id of each topic using one pipelined round trip per db_shard + def stream_heads(topics, item_objects_by_topic, packet_objects_by_topic) + heads = {} + # Every object on a topic lives on the same db_shard so the first one is enough + topics_by_shard = topics.group_by do |topic| + # Removing the last item subscription leaves an empty array behind, so + # fall through to the packet subscriptions rather than defaulting to shard 0 + objects = item_objects_by_topic[topic].presence || packet_objects_by_topic[topic] + objects&.first&.db_shard || 0 + end + topics_by_shard.each do |db_shard, shard_topics| + store = OpenC3::EphemeralStore.instance(db_shard: db_shard) + # Pipeline so N topics cost a single round trip instead of N + results = store.redis_pool.pipelined do + # XREVRANGE with COUNT 1 returns just the newest entry + shard_topics.each { |topic| store.xrevrange(topic, count: 1) } + end + # Each result is [[msg_id, fields]] (or empty) so dig out the msg_id + shard_topics.zip(results) { |topic, result| heads[topic] = result&.dig(0, 0) } + end + heads + end + + # Redis stream ids are "-"; return the milliseconds + def msg_id_ms(msg_id) + msg_id.split('-')[0].to_i + end end diff --git a/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb b/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb index bd2fe62bbd..2976acc9bf 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb @@ -38,6 +38,7 @@ def initialize(subscription_key, scope: nil) @mutex = Mutex.new @realtime_thread = nil @logged_threads = [] + @notices = false end # Create new StreamingObjects using the data['items'] and add them to the StreamingObjectCollection @@ -168,6 +169,8 @@ def expand_item_globs(data, scope:) # token: authorization token # start_time: 64-bit nanoseconds from unix epoch - If not present then realtime # end_time: 64-bit nanoseconds from unix epoch - If not present stream forever + # notices: true to also receive notice Hashes, e.g. { "__type" => "NOTICE", "notice" => "SKIPPED", + # "message" => "...", "skipped_seconds" => 12.3 } when realtime streaming falls behind and skips ahead # items: [ [ MODE__CMDORTLM__TARGET__PACKET__ITEM__VALUETYPE__REDUCEDTYPE, item_key] ] # MODE - RAW, DECOM, REDUCED_MINUTE, REDUCED_HOUR, or REDUCED_DAY # CMDORTLM - CMD or TLM @@ -195,6 +198,9 @@ def add(data) end_time = data["end_time"].to_i if data["end_time"] scope = data["scope"] token = data["token"] + # Clients opt in to out-of-band notices (a Hash rather than the usual + # Array of results) so older clients never see a message they can't parse + @notices = true if data["notices"] # Expand ALL wildcards in packets before building the collection expand_all_packets(data, scope: scope) @@ -334,6 +340,14 @@ def transmit_results(results, force: false) end end + # Send an out-of-band notice (e.g. that data was skipped) to clients that opted in + def transmit_notice(notice, message, **extra) + return unless @notices + # Notices are a Hash while normal results are an Array, which is how clients + # tell them apart. Extra keys are stringified to match the JSON the client sees. + ActionCable.server.broadcast(@subscription_key, { "__type" => "NOTICE", "notice" => notice, "message" => message }.merge(extra.transform_keys(&:to_s))) + end + # Returns if the calling thread should be canceled or not def handoff_to_realtime(collection) @mutex.synchronize do diff --git a/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb b/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb index f865a436ac..cd45c466b4 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb @@ -102,9 +102,12 @@ def redis_thread_body # Read from each db_shard with proportionally shorter timeouts timeout_per_db_shard = [500 / [db_shard_groups.length, 1].max, 100].max any_result = false + newest_msg_ids = {} # topic => newest msg_id read this pass, used to measure lag db_shard_groups.each do |db_shard, group| break if @cancel_thread xread_result = OpenC3::Topic.read_topics(group[:topics], group[:offsets], timeout_per_db_shard, @max_batch_size, db_shard: db_shard) do |topic, msg_id, msg_hash, _| + # Recorded before the skip check so skipped messages still count as read + newest_msg_ids[topic] = msg_id if skip_message?(topic, msg_hash) # Still advance offsets past skipped messages, otherwise a run of them # longer than the read count would be re-read forever @@ -165,11 +168,21 @@ def redis_thread_body # Check for completed objects by wall clock time if we got nothing check_for_completed_objects() unless any_result + + # Done after processing so a skip only affects where the next pass reads from + check_lag(newest_msg_ids, item_objects_by_topic, packet_objects_by_topic) unless @cancel_thread else @cancel_thread = true end end + # Called after each pass through Redis with the newest msg_id read per topic. + # Subclasses that stream realtime data override this to react to falling behind. + def check_lag(newest_msg_ids, item_objects_by_topic, packet_objects_by_topic) + # Intentionally empty: historical (logged) streams replay old data on + # purpose so being behind realtime is expected and not acted upon + end + # Returns true if a streamed message should not be delivered def skip_message?(_topic, msg_hash) # Ignore stored packets while realtime streaming diff --git a/openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb new file mode 100644 index 0000000000..86691b0ed8 --- /dev/null +++ b/openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb @@ -0,0 +1,118 @@ +# encoding: ascii-8bit + +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +require "rails_helper" + +RSpec.describe ApplicationCable::BroadcasterRegistry do + let(:registry) { described_class } + let(:broadcaster) { double("broadcaster", kill: nil) } + + def now + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + before(:each) do + registry.clear + allow(OpenC3::Logger).to receive(:warn) + allow(OpenC3::Logger).to receive(:error) + end + after(:each) { registry.clear } + + it "registers, gets and unregisters" do + registry.register('key', broadcaster) + expect(registry.get('key')).to equal(broadcaster) + expect(broadcaster).to receive(:kill) + expect(registry.unregister('key')).to be true + expect(registry.get('key')).to be_nil + expect(registry.unregister('key')).to be false + end + + it "touch returns false for unknown keys" do + expect(registry.touch('missing', heartbeat: true)).to be false + end + + it "keeps heartbeating entries alive while they heartbeat" do + registry.register('key', broadcaster) + registry.touch('key', heartbeat: true) + expect(registry.sweep(now + registry.ttl - 5)).to be_empty + expect(registry.registered?('key')).to be true + end + + it "reaps only heartbeating entries past the ttl" do + legacy = double("legacy", kill: nil) + registry.register('key', broadcaster) + registry.register('legacy', legacy) + registry.touch('key', heartbeat: true) + expect(broadcaster).to receive(:kill) + expect(legacy).not_to receive(:kill) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['key'] + expect(registry.registered?('legacy')).to be true + end + + it "reaps an entry registered as heartbeating before its first heartbeat" do + registry.register('key', broadcaster, heartbeat: true) + expect(broadcaster).to receive(:kill) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['key'] + end + + it "only unregisters the registration matching the token" do + newer = double("newer", kill: nil) + registry.register('key', broadcaster, token: 'old') + registry.register('key', newer, token: 'new') + expect(registry.unregister('key', token: 'old')).to be false + expect(registry.get('key')).to equal(newer) + expect(newer).not_to have_received(:kill) + expect(registry.unregister('key', token: 'new')).to be true + expect(newer).to have_received(:kill) + expect(registry.registered?('key')).to be false + end + + context "environment overrides" do + def stub_env(name, value) + allow(ENV).to receive(:fetch).and_call_original + allow(ENV).to receive(:fetch).with(name, anything).and_return(value) + end + + it "honors OPENC3_CABLE_SUBSCRIPTION_TTL" do + stub_env('OPENC3_CABLE_SUBSCRIPTION_TTL', '200') + registry.register('key', broadcaster) + registry.touch('key', heartbeat: true) + expect(registry.sweep(now + 201)).to eql ['key'] + end + + it "clamps a ttl below the minimum and warns once" do + stub_env('OPENC3_CABLE_SUBSCRIPTION_TTL', '10') + expect(OpenC3::Logger).to receive(:warn).with(/OPENC3_CABLE_SUBSCRIPTION_TTL=10/).once + expect(registry.ttl).to eql described_class::MIN_TTL.to_f + expect(registry.ttl).to eql described_class::MIN_TTL.to_f + end + + it "uses the default ttl when the value is not a number" do + stub_env('OPENC3_CABLE_SUBSCRIPTION_TTL', 'abc') + expect(registry.ttl).to eql described_class::DEFAULT_TTL.to_f + expect(OpenC3::Logger).to have_received(:warn).with(/OPENC3_CABLE_SUBSCRIPTION_TTL=abc/) + end + + it "clamps a zero sweep interval" do + stub_env('OPENC3_CABLE_SWEEP_INTERVAL', '0') + expect(registry.sweep_interval).to eql described_class::MIN_SWEEP_INTERVAL.to_f + end + end + + it "survives a broadcaster whose kill raises" do + allow(broadcaster).to receive(:kill).and_raise("boom") + registry.register('key', broadcaster) + registry.touch('key', heartbeat: true) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['key'] + end +end diff --git a/openc3-cosmos-cmd-tlm-api/spec/channels/streaming_channel_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/channels/streaming_channel_spec.rb index 02a666e337..9b03a78aef 100644 --- a/openc3-cosmos-cmd-tlm-api/spec/channels/streaming_channel_spec.rb +++ b/openc3-cosmos-cmd-tlm-api/spec/channels/streaming_channel_spec.rb @@ -80,7 +80,7 @@ def pubsub # client sends, so validate_data passes and we actually exercise the guard.) before do subscribe() - StreamingChannel.class_variable_get(:@@broadcasters)['streaming_12345'] = nil + ApplicationCable::BroadcasterRegistry.unregister('streaming_12345') end it "does not reject a remove" do @@ -93,4 +93,75 @@ def pubsub expect(subscription).not_to be_rejected end end + + context "broadcaster lifecycle" do + let(:registry) { ApplicationCable::BroadcasterRegistry } + + before(:each) { allow(OpenC3::Logger).to receive(:warn) } + + it "registers on subscribe and unregisters on unsubscribe" do + subscribe() + expect(registry.get('streaming_12345')).to be_a(StreamingApi) + unsubscribe() + expect(registry.registered?('streaming_12345')).to be false + end + + it "kills a previous broadcaster registered under the same key" do + subscribe() + first = registry.get('streaming_12345') + expect(first).to receive(:kill) + subscribe() + expect(registry.get('streaming_12345')).not_to equal(first) + end + + it "reaps a heartbeating subscription that stops heartbeating" do + subscribe() + subscription.heartbeat({}) + broadcaster = registry.get('streaming_12345') + expect(broadcaster).to receive(:kill) + now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['streaming_12345'] + expect(registry.registered?('streaming_12345')).to be false + end + + it "does not reap a subscription that never heartbeats" do + subscribe() + now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + expect(registry.sweep(now + registry.ttl + 1)).to be_empty + expect(registry.registered?('streaming_12345')).to be true + end + + it "reaps a subscription that said it heartbeats before its first heartbeat" do + subscribe(heartbeat: true) + expect(registry.get('streaming_12345')).to receive(:kill) + now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['streaming_12345'] + end + + it "does not let a late unsubscribe remove a newer subscription's broadcaster" do + subscribe() + first = subscription + allow(registry.get('streaming_12345')).to receive(:kill) + subscribe() + second = registry.get('streaming_12345') + expect(second).not_to receive(:kill) + first.unsubscribe_from_channel + expect(registry.get('streaming_12345')).to equal(second) + end + + it "does nothing on unsubscribe when subscribe was rejected" do + subscribe() + subscription.broadcaster_token = nil + expect(registry.get('streaming_12345')).not_to receive(:kill) + unsubscribe() + expect(registry.registered?('streaming_12345')).to be true + end + + it "tells a heartbeating client to resubscribe once its broadcaster is gone" do + subscribe() + registry.unregister('streaming_12345') + subscription.heartbeat({}) + expect(transmissions.last).to eql({ '__openc3_cable__' => 'resubscribe' }) + end + end end diff --git a/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb new file mode 100644 index 0000000000..d70ad12d5b --- /dev/null +++ b/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb @@ -0,0 +1,154 @@ +# encoding: ascii-8bit + +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +require 'rails_helper' + +RSpec.describe RealtimeStreamingThread, type: :model do + let(:topic) { 'DEFAULT__DECOM__{INST}__HEALTH_STATUS' } + let(:other) { 'DEFAULT__DECOM__{INST}__ADCS' } + # Arbitrary Redis server time (ms) that msg_ids are stamped with + let(:base_ms) { 1_700_000_000_000 } + + before(:each) do + @streaming_api = double('StreamingApi') + allow(@streaming_api).to receive(:transmit_notice) + @object = double('StreamingObject', db_shard: 0) + allow(@object).to receive(:offset=) + @other_object = double('OtherStreamingObject', db_shard: 0) + allow(@other_object).to receive(:offset=) + @thread = RealtimeStreamingThread.new(@streaming_api, StreamingObjectCollection.new, scope: 'DEFAULT') + allow(Process).to receive(:clock_gettime).and_call_original + end + + def at(seconds) + allow(Process).to receive(:clock_gettime).with(Process::CLOCK_MONOTONIC).and_return(1000.0 + seconds) + end + + def head_id(lag_seconds) + "#{base_ms + (lag_seconds * 1000).to_i}-0" + end + + # Simulate a pass where the newest message read trails the stream head by lag_seconds + def check(lag_seconds, seconds) + at(seconds) + allow(@thread).to receive(:stream_heads).and_return({ topic => head_id(lag_seconds) }) + @thread.check_lag({ topic => "#{base_ms}-0" }, { topic => [@object] }, {}) + end + + it "does nothing while within the max lag" do + check(1, 0) + check(9, 20) + expect(@streaming_api).to_not have_received(:transmit_notice) + expect(@object).to_not have_received(:offset=) + end + + it "skips ahead to the stream head when the lag doesn't shrink over a window" do + check(15, 0) + expect(@streaming_api).to_not have_received(:transmit_notice) + check(16, 5) + expect(@streaming_api).to_not have_received(:transmit_notice) + check(20, 11) + expect(@object).to have_received(:offset=).with(head_id(20)) + expect(@streaming_api).to have_received(:transmit_notice).with('SKIPPED', /skipped 20.0s/, skipped_seconds: be_within(0.01).of(20.0)) + end + + it "gives a stream that is catching up more time" do + check(40, 0) + check(30, 11) + check(20, 22) + check(11, 33) + expect(@streaming_api).to_not have_received(:transmit_notice) + check(5, 44) + check(15, 50) + check(15, 55) + expect(@streaming_api).to_not have_received(:transmit_notice) + end + + it "skips immediately when far behind" do + check(61, 0) + expect(@streaming_api).to have_received(:transmit_notice).with('SKIPPED', anything, skipped_seconds: be_within(0.01).of(61.0)) + end + + it "only checks the lag once per interval" do + check(1, 0) + check(70, 0.5) + expect(@thread).to have_received(:stream_heads).once + expect(@streaming_api).to_not have_received(:transmit_notice) + check(70, 1.5) + expect(@streaming_api).to have_received(:transmit_notice) + end + + it "uses the most behind topic and only advances topics that are behind" do + at(0) + allow(@thread).to receive(:stream_heads).and_return({ topic => head_id(1), other => head_id(70) }) + @thread.check_lag({ topic => "#{base_ms}-0", other => "#{base_ms}-0" }, { topic => [@object] }, { other => [@other_object] }) + expect(@other_object).to have_received(:offset=).with(head_id(70)) + expect(@object).to_not have_received(:offset=) + expect(@streaming_api).to have_received(:transmit_notice).with('SKIPPED', anything, skipped_seconds: be_within(0.01).of(70.0)) + end + + it "can be disabled" do + stub_const('ENV', ENV.to_h.merge('OPENC3_STREAMING_MAX_LAG_SECONDS' => '0')) + @thread = RealtimeStreamingThread.new(@streaming_api, StreamingObjectCollection.new, scope: 'DEFAULT') + allow(@thread).to receive(:stream_heads) + check(1000, 0) + expect(@thread).to_not have_received(:stream_heads) + expect(@streaming_api).to_not have_received(:transmit_notice) + end + + it "falls back to the default max lag when the setting is malformed" do + stub_const('ENV', ENV.to_h.merge('OPENC3_STREAMING_MAX_LAG_SECONDS' => 'ten')) + allow(OpenC3::Logger).to receive(:warn) + @thread = RealtimeStreamingThread.new(@streaming_api, StreamingObjectCollection.new, scope: 'DEFAULT') + expect(OpenC3::Logger).to have_received(:warn).with(/Invalid OPENC3_STREAMING_MAX_LAG_SECONDS 'ten'/) + check(61, 0) + expect(@streaming_api).to have_received(:transmit_notice).with('SKIPPED', anything, skipped_seconds: be_within(0.01).of(61.0)) + end + + it "reads the stream head from the packet subscription's shard when no item subscriptions remain" do + shard_object = double('ShardObject', db_shard: 2) + store = double('EphemeralStore') + pool = double('RedisPool') + allow(store).to receive(:redis_pool).and_return(pool) + allow(store).to receive(:xrevrange) + allow(pool).to receive(:pipelined) { |&block| block.call; [[['5000-0', {}]]] } + expect(OpenC3::EphemeralStore).to receive(:instance).with(db_shard: 2).and_return(store) + # An empty item array is what StreamingObjectCollection#remove leaves behind + heads = @thread.stream_heads([topic], { topic => [] }, { topic => [shard_object] }) + expect(heads).to eql({ topic => '5000-0' }) + end + + context "with redis" do + before(:each) do + mock_redis() + end + + it "reads the newest msg_id of each topic" do + OpenC3::Topic.write_topic(topic, { 'a' => '1' }, '1000-0') + OpenC3::Topic.write_topic(topic, { 'a' => '2' }, '2000-0') + OpenC3::Topic.write_topic(other, { 'a' => '3' }, '3000-0') + heads = @thread.stream_heads([topic, other, 'DEFAULT__DECOM__{INST}__EMPTY'], { topic => [@object] }, { other => [@object] }) + expect(heads).to eql({ topic => '2000-0', other => '3000-0', 'DEFAULT__DECOM__{INST}__EMPTY' => nil }) + end + + it "is not fooled by clock skew between this host and redis" do + msg_id = OpenC3::Topic.write_topic(topic, { 'a' => '1' }) + # This host's clock is far ahead of the Redis server that stamped msg_id + allow(Time).to receive(:now).and_return(Time.now + 120) + at(0) + @thread.check_lag({ topic => msg_id }, { topic => [@object] }, {}) + expect(@streaming_api).to_not have_received(:transmit_notice) + expect(@object).to_not have_received(:offset=) + end + end +end diff --git a/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb index b5ecc03e57..1b4d424d17 100644 --- a/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb +++ b/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb @@ -143,6 +143,23 @@ def response.is_truncated expect(@api.instance_variable_get('@logged_threads')).to be_empty end + describe '#transmit_notice' do + it 'only sends notices to clients that opted in' do + @api.transmit_notice('SKIPPED', 'skipped', skipped_seconds: 1.5) + expect(@messages).to be_empty + + @api.instance_variable_set('@notices', true) + @api.transmit_notice('SKIPPED', 'skipped', skipped_seconds: 1.5) + expect(@messages).to eq([{ "__type" => "NOTICE", "notice" => "SKIPPED", "message" => "skipped", "skipped_seconds" => 1.5 }]) + end + + it 'opts in via the add request' do + allow(StreamingObject).to receive(:new).and_raise(OpenC3::ForbiddenError) + @api.add({ 'scope' => 'DEFAULT', 'notices' => true, 'end_time' => 0, 'packets' => ['DECOM__TLM__INST__PARAMS__CONVERTED'] }) + expect(@api.instance_variable_get('@notices')).to be true + end + end + describe '#kill' do class FakeStreamingThread < StreamingThread def initialize(streaming_api, stop_delay) @@ -653,6 +670,23 @@ def start_fake_thread(stop_delay) @api.kill expect(@api.instance_variable_get('@realtime_thread')).to be_nil end + + it 'skips ahead and notifies when far behind realtime' do + # Messages read are 2 minutes older than the stream head so we're well past the max lag + @time = @start_time - 120 + head = "#{Time.now.to_i * 1000}-0" + allow_any_instance_of(RealtimeStreamingThread).to receive(:stream_heads) do |_thread, topics| + topics.to_h { |topic| [topic, head] } + end + @send_count = 1 + data['notices'] = true + @api.add(data) + sleep 0.35 # Allow the thread to run + notice = @messages.find { |message| message.is_a?(Hash) } + expect(notice['notice']).to eq('SKIPPED') + expect(notice['skipped_seconds']).to be > 60 + @api.kill + end end context 'logging plus realtime' do diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/cmd_tlm/inst_cmds.txt b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/cmd_tlm/inst_cmds.txt index 7f7214c3f9..47e243afaf 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/cmd_tlm/inst_cmds.txt +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/cmd_tlm/inst_cmds.txt @@ -279,6 +279,19 @@ COMMAND <%= target_name %> BLOCK BIG_ENDIAN "Send variable block of data" HIDDEN RELATED_ITEM <%= target_name %> HEALTH_STATUS BLOCKTEST +COMMAND <%= target_name %> MEMDUMP BIG_ENDIAN "Dump a block of simulated memory" + <%= render "_ccsds_cmd.txt", locals: {id: 17} %> + APPEND_PARAMETER OFFSET 32 UINT MIN MAX 0 "Memory offset in bytes to start the dump" + APPEND_PARAMETER SIZE 32 UINT 1 4096 1024 "Number of bytes to dump" + UNITS Bytes B + APPEND_PARAMETER BAD_CRC 8 UINT 0 1 0 "Flip a random bit in the response DATA after the CRC is calculated to simulate a bad CRC" + STATE FALSE 0 + STATE TRUE 1 + APPEND_PARAMETER CRC 32 UINT MIN MAX 0 "CRC32 over OFFSET, SIZE and BAD_CRC (calculated automatically, use cmd_raw to override)" + FORMAT_STRING "0x%08X" + WRITE_CONVERSION memdump_cmd_crc_conversion.rb + RELATED_ITEM <%= target_name %> MEMDUMP DATA + # Throughput Testing Commands (for use with external throughput server) COMMAND <%= target_name %> START_STREAM BIG_ENDIAN "Start telemetry streaming from throughput server" <%= render "_ccsds_cmd.txt", locals: {id: 200} %> diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/cmd_tlm/inst_tlm.txt b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/cmd_tlm/inst_tlm.txt index 5ebcc75b0a..ad9975b227 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/cmd_tlm/inst_tlm.txt +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/cmd_tlm/inst_tlm.txt @@ -215,6 +215,30 @@ TELEMETRY <%= target_name %> VARIABLE_ARRAYS BIG_ENDIAN "Variable length arrays" READ_CONVERSION openc3/conversions/unix_time_conversion.py TIMESEC TIMEUS CONVERTED_DATA 280 STRING +TELEMETRY <%= target_name %> MEMDUMP BIG_ENDIAN "Memory dump sent in response to the MEMDUMP command" + <%= render "_ccsds_tlm.txt", locals: {apid: 8} %> + APPEND_ITEM OFFSET 32 UINT "Memory offset in bytes of the dump" + APPEND_ITEM SIZE 32 UINT "Number of bytes in the dump" + UNITS Bytes B + APPEND_ITEM DATA 0 BLOCK "Dumped memory" + VARIABLE_BIT_SIZE SIZE + APPEND_ITEM CRC 32 UINT "CRC32 of DATA calculated by the target" + FORMAT_STRING "0x%08X" + ITEM CALC_CRC 0 0 DERIVED "CRC32 of DATA calculated on the ground" + READ_CONVERSION crc32_conversion.rb DATA + CONVERTED_DATA 32 UINT + FORMAT_STRING "0x%08X" + ITEM CRC_OK 0 0 DERIVED "Whether the target CRC matches the ground CRC" + GENERIC_READ_CONVERSION_START + packet.read('CRC') == packet.read('CALC_CRC') ? 1 : 0 + GENERIC_READ_CONVERSION_END + CONVERTED_DATA 8 UINT + STATE TRUE 1 GREEN + STATE FALSE 0 RED + ITEM PACKET_TIME 0 0 DERIVED "Ruby time based on TIMESEC and TIMEUS" + READ_CONVERSION unix_time_conversion.rb TIMESEC TIMEUS + CONVERTED_DATA 280 STRING + TELEMETRY <%= target_name %> JSONTLM BIG_ENDIAN "JSON Accessor Telemetry" ACCESSOR JsonAccessor # Template is not required for telemetry, but is useful for simulation diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/crc32_conversion.rb b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/crc32_conversion.rb new file mode 100644 index 0000000000..8d42c2e376 --- /dev/null +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/crc32_conversion.rb @@ -0,0 +1,38 @@ +# encoding: ascii-8bit + +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +require 'openc3/conversions/conversion' +require 'openc3/utilities/crc' + +module OpenC3 + # Read conversion which calculates the CRC32 of another item in the packet. + # Used to verify the CRC sent by the target. + # See https://docs.openc3.com/docs/configuration/conversions + class Crc32Conversion < Conversion + # @param item_name [String] Name of the item to calculate the CRC over + def initialize(item_name) + super() + @item_name = item_name.to_s.upcase + @crc = Crc32.new + @params = [@item_name] + end + + # @param value [Object] Value based on the item definition. This could be + # a string, integer, float, or array of values. + # @param packet [Packet] The packet object where the conversion is defined + # @param buffer [String] The raw packet buffer + def call(_value, packet, buffer) + @crc.calc(packet.read(@item_name, :RAW, buffer)) + end + end +end diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/memdump_cmd_crc_conversion.rb b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/memdump_cmd_crc_conversion.rb new file mode 100644 index 0000000000..e169a2417c --- /dev/null +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/memdump_cmd_crc_conversion.rb @@ -0,0 +1,46 @@ +# encoding: ascii-8bit + +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +require 'openc3/conversions/conversion' +require 'openc3/utilities/crc' + +module OpenC3 + # Write conversion which calculates the MEMDUMP command CRC32 over the + # big endian OFFSET, SIZE and BAD_CRC parameters so the user never has to. + # Send the command with cmd_raw to bypass this conversion and set a bad CRC. + # See https://docs.openc3.com/docs/configuration/conversions + class MemdumpCmdCrcConversion < Conversion + def initialize + super() + @crc = Crc32.new + end + + # @param value [Object] Value based on the item definition. This could be + # a string, integer, float, or array of values. + # @param packet [Packet] The packet object where the conversion is defined + # @param buffer [String] The raw packet buffer + def call(_value, packet, _buffer) + # Write conversions run before the given parameters are written to the + # buffer so use the given values and fall back to the defaults. + # Given names keep the caller's case so upcase them to match the items. + given = (packet.given_values || {}).transform_keys { |key| key.to_s.upcase } + offset = Integer(given['OFFSET'] || packet.get_item('OFFSET').default) + size = Integer(given['SIZE'] || packet.get_item('SIZE').default) + # BAD_CRC may be given as a state name so convert it to its value + bad_crc_item = packet.get_item('BAD_CRC') + bad_crc = given['BAD_CRC'] || bad_crc_item.default + bad_crc = bad_crc_item.states[bad_crc.to_s.upcase] || bad_crc + @crc.calc([offset, size, Integer(bad_crc)].pack('NNC')) + end + end +end diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/sim_inst.rb b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/sim_inst.rb index 317f73cbdd..b91329989c 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/sim_inst.rb +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/lib/sim_inst.rb @@ -21,6 +21,7 @@ require 'stringio' require 'base64' require 'openc3/accessors/binary_accessor' +require 'openc3/utilities/crc' module OpenC3 # Simulated instrument for the demo. Populates several packets and cycles @@ -128,6 +129,10 @@ def initialize(target_name) packet.CcsdsSeqFlags = 'NOGROUP' packet.CcsdsLength = packet.buffer.length - 7 + packet = @tlm_packets['MEMDUMP'] + packet.enable_method_missing + packet.CcsdsSeqFlags = 'NOGROUP' + @solar_panel_positions = SOLAR_PANEL_DFLTS.dup @solar_panel_thread = nil @solar_panel_thread_cancel = false @@ -150,6 +155,8 @@ def initialize(target_name) @time_offset = 0 @ip_address = 0 @variable_arrays_updated = false + @memdump_response = nil + @crc = Crc32.new end def set_rates @@ -269,6 +276,26 @@ def write(packet) # Write only arrays - variable_bit_size will auto-update length fields arrays_packet.write('ARRAY1', packet.read('ARRAY1')) arrays_packet.write('ARRAY2', packet.read('ARRAY2')) + when 'MEMDUMP' + offset = packet.read('OFFSET') + size = packet.read('SIZE') + bad_crc = packet.read('BAD_CRC', :RAW) + # Reject the command if the CRC over OFFSET, SIZE and BAD_CRC doesn't match + return unless packet.read('CRC') == @crc.calc([offset, size, bad_crc].pack('NNC')) + # Reject sizes outside the command's range since cmd_raw skips range checks + return unless packet.get_item('SIZE').range.cover?(size) + hs_packet.cmd_acpt_cnt += 1 + # Simulated memory is a byte counter where each byte holds its address + data = Array.new(size) { |i| (offset + i) & 0xFF }.pack('C*') + crc = @crc.calc(data) + if bad_crc == 1 + # Flip a random bit after calculating the CRC to simulate downlink corruption + bit = rand(size * 8) + data.setbyte(bit / 8, data.getbyte(bit / 8) ^ (0x80 >> (bit % 8))) + end + # Commands arrive on a different thread than read so only hand off the + # values here and let read write them into the MEMDUMP packet + @memdump_response = [offset, data, crc] end end @@ -469,6 +496,21 @@ def read(count_100hz, time) pending_packets << arrays_packet end + memdump_response, @memdump_response = @memdump_response, nil + if memdump_response + offset, data, crc = memdump_response + memdump_packet = @tlm_packets['MEMDUMP'] + memdump_packet.write('OFFSET', offset) + # Writing DATA resizes the packet and updates SIZE due to VARIABLE_BIT_SIZE + memdump_packet.write('DATA', data) + memdump_packet.write('CRC', crc) + memdump_packet.write('CCSDSLENGTH', memdump_packet.buffer(false).length - 7) + memdump_packet.write('TIMESEC', time.tv_sec - @time_offset) + memdump_packet.write('TIMEUS', time.tv_usec) + memdump_packet.write('CCSDSSEQCNT', memdump_packet.read('CCSDSSEQCNT') + 1) + pending_packets << memdump_packet + end + pending_packets end end diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/procedures/memdump.rb b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/procedures/memdump.rb new file mode 100644 index 0000000000..3eb4baa0fb --- /dev/null +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/procedures/memdump.rb @@ -0,0 +1,130 @@ +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +# Dump a region of INST memory with the MEMDUMP command, verify each chunk +# and then push the dump to the bucket with put_target_file. +# +# The dump recovers from a bad downlink: +# * Chunks with no response are requested again +# * Chunks with a bad CRC prompt the operator to retry or stop the dump +# * If the link is down the script waits for it to come back +# +# Environment variables: +# BAD_CRC=1 flips a bit in one chunk part way through the dump to show recovery from a bad CRC +require 'openc3/utilities/crc' + +START_OFFSET = 0 +TOTAL_SIZE = 10000 # Bytes to dump +CHUNK_SIZE = 1024 # Bytes per MEMDUMP command (max 4096) +MAX_ATTEMPTS = 3 # Times to request a chunk with no response before giving up +RESPONSE_TIMEOUT = 5 # Seconds to wait for each MEMDUMP response +LINK_TIMEOUT = 60 # Seconds to wait for a lost link to come back + +crc = OpenC3::Crc32.new +bad_crc = ENV.fetch('BAD_CRC', nil) +end_offset = START_OFFSET + TOTAL_SIZE +dump_data = ''.b # The verified dump is small enough to keep in memory +offset = START_OFFSET +bucket_path = "INST/memdumps/INST_memdump_#{Time.now.utc.strftime('%Y%m%d%H%M%S')}.bin" + +bad_crc_offset = START_OFFSET + (TOTAL_SIZE / CHUNK_SIZE / 2) * CHUNK_SIZE +while offset < end_offset + size = [CHUNK_SIZE, end_offset - offset].min + # Request a bad CRC on the first request for the chunk in the middle of + # the dump + chunk_bad_crc = 'FALSE' + if bad_crc and offset >= bad_crc_offset + chunk_bad_crc = 'TRUE' + bad_crc = nil + end + + data = nil + attempt = 0 + while data.nil? and attempt < MAX_ATTEMPTS + attempt += 1 + # Record the received count so we know when the response arrives + count = tlm("INST MEMDUMP RECEIVED_COUNT") || 0 + + # Send the command, waiting for the link if it is down. This doesn't use up + # an attempt. An ack timeout may mean the command did reach the target but + # the duplicate response is harmless because it has the same OFFSET and SIZE. + link_deadline = Time.now + LINK_TIMEOUT + sent = false + until sent + begin + # The command CRC over OFFSET, SIZE and BAD_CRC is calculated automatically + cmd("INST MEMDUMP with OFFSET #{offset}, SIZE #{size}, BAD_CRC #{chunk_bad_crc}") + sent = true + # Only the first request has a bad CRC so the retry succeeds + chunk_bad_crc = 'FALSE' + rescue => e + raise unless e.message =~ /not connected|waiting for cmd ack/i and Time.now < link_deadline + puts "WARN: #{e.message}, retrying until the link returns" + wait(2) + end + end + + # Wait for the response to this request. A stale response (e.g. a late + # reply to an earlier request) is skipped and we keep waiting. + dump = nil + deadline = Time.now + RESPONSE_TIMEOUT + while (remaining = deadline - Time.now) > 0 + # wait returns false on timeout rather than raising + break unless wait("INST MEMDUMP RECEIVED_COUNT > #{count}", remaining) + # Grab all the items at once so they all come from the same packet + response = get_tlm_packet("INST MEMDUMP", type: :RAW).to_h { |name, value, _| [name, value] } + count = response['RECEIVED_COUNT'] + if response['OFFSET'] == offset and response['SIZE'] == size + dump = response + break + end + puts "WARN: Ignoring stale response for OFFSET #{response['OFFSET']}, SIZE #{response['SIZE']} " \ + "while waiting for OFFSET #{offset}, SIZE #{size}" + end + unless dump + puts "WARN: No response for OFFSET #{offset} (attempt #{attempt} of #{MAX_ATTEMPTS})" + next + end + + # Calculate the CRC in the script (the CRC_OK derived item does the same + # check on the ground for display on the screen) + calculated = crc.calc(dump['DATA']) + if calculated == dump['CRC'] + puts "CRC passed at OFFSET #{offset}" + data = dump['DATA'] + else + mismatch = "Target CRC 0x#{dump['CRC'].to_s(16).upcase} vs calculated 0x#{calculated.to_s(16).upcase}" + puts "WARN: CRC mismatch at OFFSET #{offset}: #{mismatch}" + # Let the operator decide whether to request the chunk again + answer = message_box("Bad CRC at OFFSET #{offset}. Retry this chunk?", 'Retry', 'Stop Dump', informative: mismatch) + if answer == 'Stop Dump' + raise "Dump stopped by operator at OFFSET #{offset}" + end + # An operator retry doesn't use up an attempt + attempt -= 1 + end + end + raise "Unable to dump OFFSET #{offset} after #{MAX_ATTEMPTS} attempts" unless data + dump_data << data.b + offset += size +end +puts "Dumped #{dump_data.length} bytes (CRC32 0x#{crc.calc(dump_data).to_s(16).upcase})" + +# Push the dump to the bucket +put_target_file(bucket_path, dump_data) + +# Read it back to verify it made it to the bucket intact +file = get_target_file(bucket_path) +data = file.read +file.delete +check_expression("#{data.length} == #{TOTAL_SIZE}") +check_expression("#{crc.calc(data)} == #{crc.calc(Array.new(TOTAL_SIZE) { |i| (START_OFFSET + i) & 0xFF }.pack('C*'))}") +puts "Memory dump saved to #{bucket_path}" diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/screens/memdump.txt b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/screens/memdump.txt new file mode 100644 index 0000000000..beb78a42e6 --- /dev/null +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST/screens/memdump.txt @@ -0,0 +1,40 @@ +SCREEN AUTO AUTO 1.0 + +TITLE "<%= target_name %> Memory Dump" + +HORIZONTAL + VERTICALBOX "Commands" + MATRIXBYCOLUMNS 2 + LABEL "Offset:" + NAMED_WIDGET OFFSET TEXTFIELD 10 "0" + LABEL "Size:" + NAMED_WIDGET SIZE TEXTFIELD 10 "256" + END + # Flips a random bit in the dump after the target calculates its CRC + NAMED_WIDGET BAD_CRC CHECKBUTTON 'Bad CRC' + # The command CRC over OFFSET, SIZE and BAD_CRC is calculated automatically + BUTTON 'Dump Memory' "var offset=screen.getNamedWidget('OFFSET').text();" + + "var size=screen.getNamedWidget('SIZE').text();" + + "var badCrc=screen.getNamedWidget('BAD_CRC').checked() ? 'TRUE' : 'FALSE';" + + "api.cmd('<%= target_name %> MEMDUMP with OFFSET '+offset+', SIZE '+size+', BAD_CRC '+badCrc)" + # cmd_raw skips the CRC write conversion so the target rejects the command + BUTTON 'Run Dump Script' "runScript('<%= target_name %>/procedures/memdump.rb', true)" + BUTTON 'Run Dump Script with Bad CRC' "runScript('<%= target_name %>/procedures/memdump.rb', true, {'BAD_CRC': '1'})" + END + + VERTICALBOX "Last Dump" + LABELVALUE <%= target_name %> MEMDUMP OFFSET + LABELVALUE <%= target_name %> MEMDUMP SIZE + LABELVALUE <%= target_name %> MEMDUMP CRC + LABELVALUE <%= target_name %> MEMDUMP CALC_CRC + LABELLED <%= target_name %> MEMDUMP CRC_OK + SETTING LED_COLOR TRUE GREEN + SETTING LED_COLOR FALSE RED + LABELVALUE <%= target_name %> MEMDUMP RECEIVED_COUNT + LABELVALUE <%= target_name %> MEMDUMP PACKET_TIME + END +END + +VERTICALBOX "Data" + BLOCK <%= target_name %> MEMDUMP DATA 680 200 "%02X" 4 8 "0x%08X: " +END diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/cmd_tlm/inst_cmds.txt b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/cmd_tlm/inst_cmds.txt index 8821e300cd..ff2f336167 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/cmd_tlm/inst_cmds.txt +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/cmd_tlm/inst_cmds.txt @@ -279,6 +279,19 @@ COMMAND <%= target_name %> BLOCK BIG_ENDIAN "Send variable block of data" HIDDEN RELATED_ITEM <%= target_name %> HEALTH_STATUS BLOCKTEST +COMMAND <%= target_name %> MEMDUMP BIG_ENDIAN "Dump a block of simulated memory" + <%= render "_ccsds_cmd.txt", locals: {id: 17} %> + APPEND_PARAMETER OFFSET 32 UINT MIN MAX 0 "Memory offset in bytes to start the dump" + APPEND_PARAMETER SIZE 32 UINT 1 4096 1024 "Number of bytes to dump" + UNITS Bytes B + APPEND_PARAMETER BAD_CRC 8 UINT 0 1 0 "Flip a random bit in the response DATA after the CRC is calculated to simulate a bad CRC" + STATE FALSE 0 + STATE TRUE 1 + APPEND_PARAMETER CRC 32 UINT MIN MAX 0 "CRC32 over OFFSET, SIZE and BAD_CRC (calculated automatically, use cmd_raw to override)" + FORMAT_STRING "0x%08X" + WRITE_CONVERSION memdump_cmd_crc_conversion.py + RELATED_ITEM <%= target_name %> MEMDUMP DATA + # Throughput Testing Commands (for use with external throughput server) COMMAND <%= target_name %> START_STREAM BIG_ENDIAN "Start telemetry streaming from throughput server" <%= render "_ccsds_cmd.txt", locals: {id: 200} %> diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/cmd_tlm/inst_tlm.txt b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/cmd_tlm/inst_tlm.txt index 6206f78d7c..8d42206180 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/cmd_tlm/inst_tlm.txt +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/cmd_tlm/inst_tlm.txt @@ -208,6 +208,30 @@ TELEMETRY <%= target_name %> VARIABLE_ARRAYS BIG_ENDIAN "Variable length arrays" READ_CONVERSION openc3/conversions/unix_time_conversion.py TIMESEC TIMEUS CONVERTED_DATA 280 STRING +TELEMETRY <%= target_name %> MEMDUMP BIG_ENDIAN "Memory dump sent in response to the MEMDUMP command" + <%= render "_ccsds_tlm.txt", locals: {apid: 8} %> + APPEND_ITEM OFFSET 32 UINT "Memory offset in bytes of the dump" + APPEND_ITEM SIZE 32 UINT "Number of bytes in the dump" + UNITS Bytes B + APPEND_ITEM DATA 0 BLOCK "Dumped memory" + VARIABLE_BIT_SIZE SIZE + APPEND_ITEM CRC 32 UINT "CRC32 of DATA calculated by the target" + FORMAT_STRING "0x%08X" + ITEM CALC_CRC 0 0 DERIVED "CRC32 of DATA calculated on the ground" + READ_CONVERSION crc32_conversion.py DATA + CONVERTED_DATA 32 UINT + FORMAT_STRING "0x%08X" + ITEM CRC_OK 0 0 DERIVED "Whether the target CRC matches the ground CRC" + GENERIC_READ_CONVERSION_START + 1 if packet.read('CRC') == packet.read('CALC_CRC') else 0 + GENERIC_READ_CONVERSION_END + CONVERTED_DATA 8 UINT + STATE TRUE 1 GREEN + STATE FALSE 0 RED + ITEM PACKET_TIME 0 0 DERIVED "Python time based on TIMESEC and TIMEUS" + READ_CONVERSION openc3/conversions/unix_time_conversion.py TIMESEC TIMEUS + CONVERTED_DATA 280 STRING + TELEMETRY <%= target_name %> JSONTLM BIG_ENDIAN "JSON Accessor Telemetry" ACCESSOR JsonAccessor # Template is not required for telemetry, but is useful for simulation diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/crc32_conversion.py b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/crc32_conversion.py new file mode 100644 index 0000000000..0c73e44f85 --- /dev/null +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/crc32_conversion.py @@ -0,0 +1,32 @@ +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +from openc3.conversions.conversion import Conversion +from openc3.utilities.crc import Crc32 + + +# Read conversion which calculates the CRC32 of another item in the packet. +# Used to verify the CRC sent by the target. +# See https://docs.openc3.com/docs/configuration/conversions +class Crc32Conversion(Conversion): + # @param item_name [String] Name of the item to calculate the CRC over + def __init__(self, item_name): + super().__init__() + self.item_name = item_name.upper() + self.crc = Crc32() + self.params = [self.item_name] + + # @param value [Object] Value based on the item definition. This could be + # a string, integer, float, or array of values. + # @param packet [Packet] The packet object where the conversion is defined + # @param buffer [String] The raw packet buffer + def call(self, value, packet, buffer): + return self.crc.calc(packet.read(self.item_name, "RAW", buffer)) diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/memdump_cmd_crc_conversion.py b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/memdump_cmd_crc_conversion.py new file mode 100644 index 0000000000..502856e9c8 --- /dev/null +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/memdump_cmd_crc_conversion.py @@ -0,0 +1,42 @@ +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +import struct + +from openc3.conversions.conversion import Conversion +from openc3.utilities.crc import Crc32 + + +# Write conversion which calculates the MEMDUMP command CRC32 over the +# big endian OFFSET, SIZE and BAD_CRC parameters so the user never has to. +# Send the command with cmd_raw to bypass this conversion and set a bad CRC. +# See https://docs.openc3.com/docs/configuration/conversions +class MemdumpCmdCrcConversion(Conversion): + def __init__(self): + super().__init__() + self.crc = Crc32() + + # @param value [Object] Value based on the item definition. This could be + # a string, integer, float, or array of values. + # @param packet [Packet] The packet object where the conversion is defined + # @param buffer [String] The raw packet buffer + def call(self, value, packet, buffer): + # Write conversions run before the given parameters are written to the + # buffer so use the given values and fall back to the defaults. + # Given names keep the caller's case so upcase them to match the items. + given = {str(key).upper(): value for key, value in (packet.given_values or {}).items()} + offset = int(given.get("OFFSET", packet.get_item("OFFSET").default)) + size = int(given.get("SIZE", packet.get_item("SIZE").default)) + # BAD_CRC may be given as a state name so convert it to its value + bad_crc_item = packet.get_item("BAD_CRC") + bad_crc = given.get("BAD_CRC", bad_crc_item.default) + bad_crc = bad_crc_item.states.get(str(bad_crc).upper(), bad_crc) + return self.crc.calc(struct.pack(">IIB", offset, size, int(bad_crc))) diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/sim_inst.py b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/sim_inst.py index 0dcab4db26..3d3b4a1f9a 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/sim_inst.py +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/lib/sim_inst.py @@ -13,6 +13,7 @@ import time import base64 import random +import struct import threading from io import BytesIO from openc3.utilities.simulated_target import SimulatedTarget @@ -21,6 +22,7 @@ from openc3.system.system import System from openc3.top_level import kill_thread from openc3.accessors.binary_accessor import BinaryAccessor +from openc3.utilities.crc import Crc32 # Simulated instrument for the demo. Populates several packets and cycles @@ -124,6 +126,9 @@ def __init__(self, target_name): packet.write("CcsdsSeqFlags", "NOGROUP") packet.write("CcsdsLength", len(packet.buffer) - 7) + packet = self.tlm_packets["MEMDUMP"] + packet.write("CcsdsSeqFlags", "NOGROUP") + self.solar_panel_positions = SimInst.SOLAR_PANEL_DFLTS[:] self.solar_panel_thread = None self.solar_panel_thread_cancel = False @@ -146,6 +151,8 @@ def __init__(self, target_name): self.time_offset = 0 self.ip_address = 0 self.variable_arrays_updated = False + self.memdump_response = None + self.crc = Crc32() def set_rates(self): self.set_rate("ADCS", 10) @@ -244,6 +251,28 @@ def write(self, packet): # Write only arrays - variable_bit_size will auto-update length fields arrays_packet.write("array1", packet.read("ARRAY1")) arrays_packet.write("array2", packet.read("ARRAY2")) + case "MEMDUMP": + offset = packet.read("OFFSET") + size = packet.read("SIZE") + bad_crc = packet.read("BAD_CRC", "RAW") + # Reject the command if the CRC over OFFSET, SIZE and BAD_CRC doesn't match + if packet.read("CRC") != self.crc.calc(struct.pack(">IIB", offset, size, bad_crc)): + return + # Reject sizes outside the command's range since cmd_raw skips range checks + size_item = packet.get_item("SIZE") + if not size_item.minimum <= size <= size_item.maximum: + return + hs_packet.write("cmd_acpt_cnt", hs_packet.read("cmd_acpt_cnt") + 1) + # Simulated memory is a byte counter where each byte holds its address + data = bytearray((offset + i) & 0xFF for i in range(size)) + crc = self.crc.calc(data) + if bad_crc == 1: + # Flip a random bit after calculating the CRC to simulate downlink corruption + bit = random.randrange(size * 8) # NOSONAR - simulated bit error, not security related + data[bit // 8] ^= 0x80 >> (bit % 8) + # Commands arrive on a different thread than read so only hand off the + # values here and let read write them into the MEMDUMP packet + self.memdump_response = (offset, bytes(data), crc) def solar_panel_thread_method(self): self.solar_panel_thread_cancel = False @@ -467,4 +496,18 @@ def read(self, count_100hz, time): arrays_packet.write("ccsdsseqcnt", arrays_packet.read("ccsdsseqcnt") + 1) pending_packets.append(arrays_packet) + memdump_response, self.memdump_response = self.memdump_response, None + if memdump_response: + offset, data, crc = memdump_response + memdump_packet = self.tlm_packets["MEMDUMP"] + memdump_packet.write("OFFSET", offset) + # Writing DATA resizes the packet and updates SIZE due to VARIABLE_BIT_SIZE + memdump_packet.write("DATA", data) + memdump_packet.write("CRC", crc) + memdump_packet.write("ccsdslength", len(memdump_packet.buffer) - 7) + memdump_packet.write("timesec", int(time - self.time_offset)) + memdump_packet.write("timeus", int((time % 1) * 1000000)) + memdump_packet.write("ccsdsseqcnt", memdump_packet.read("ccsdsseqcnt") + 1) + pending_packets.append(memdump_packet) + return pending_packets diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/calendar.py b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/calendar.py index 2b21831784..aa43d8f819 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/calendar.py +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/calendar.py @@ -11,7 +11,7 @@ set_timeline_color("PythonTL", "#4287f5") print(get_timeline("PythonTL")) #=> {'name': 'PythonTL', 'color': '#4287f5', 'execute': True, 'shard': 0, 'scope': 'DEFAULT', 'updated_at': 1737129508391137136} -now = datetime.datetime.now(datetime.timezone.utc) +now = datetime.datetime.now(datetime.UTC) start = now + datetime.timedelta(hours=1) stop = start + datetime.timedelta(hours=1) act1 = create_timeline_activity("PythonTL", kind="reserve", start=start, stop=stop) @@ -45,8 +45,8 @@ # "events"=>[{"time"=>1737128761, "event"=>"created"}], "recurring"=>{}} # Update act2 with a custom title and notes -new_start = datetime.datetime.fromtimestamp(act2['start'], tz=datetime.timezone.utc) -new_stop = datetime.datetime.fromtimestamp(act2['stop'], tz=datetime.timezone.utc) + datetime.timedelta(minutes=30) +new_start = datetime.datetime.fromtimestamp(act2['start'], tz=datetime.UTC) +new_stop = datetime.datetime.fromtimestamp(act2['stop'], tz=datetime.UTC) + datetime.timedelta(minutes=30) updated = update_timeline_activity("PythonTL", id=act2['start'], kind="COMMAND", start=new_start, stop=new_stop, diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/memdump.py b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/memdump.py new file mode 100644 index 0000000000..c685bb908d --- /dev/null +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/memdump.py @@ -0,0 +1,138 @@ +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +# Dump a region of INST2 memory with the MEMDUMP command, verify each chunk +# and then push the dump to the bucket with put_target_file. +# +# The dump recovers from a bad downlink: +# * Chunks with no response are requested again +# * Chunks with a bad CRC prompt the operator to retry or stop the dump +# * If the link is down the script waits for it to come back +# +# Environment variables: +# BAD_CRC=1 flips a bit in one chunk part way through the dump to show recovery from a bad CRC +import os +import re +import time +from datetime import datetime, timezone + +from openc3.utilities.crc import Crc32 + + +START_OFFSET = 0 +TOTAL_SIZE = 10000 # Bytes to dump +CHUNK_SIZE = 1024 # Bytes per MEMDUMP command (max 4096) +MAX_ATTEMPTS = 3 # Times to request a chunk with no response before giving up +RESPONSE_TIMEOUT = 5 # Seconds to wait for each MEMDUMP response +LINK_TIMEOUT = 60 # Seconds to wait for a lost link to come back + +crc = Crc32() +bad_crc = os.environ.get("BAD_CRC") +end_offset = START_OFFSET + TOTAL_SIZE +dump_data = bytearray() # The verified dump is small enough to keep in memory +offset = START_OFFSET +bucket_path = f"INST2/memdumps/INST2_memdump_{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}.bin" + +bad_crc_offset = START_OFFSET + (TOTAL_SIZE // CHUNK_SIZE // 2) * CHUNK_SIZE +while offset < end_offset: + size = min(CHUNK_SIZE, end_offset - offset) + # Request a bad CRC on the first request for the chunk in the middle of + # the dump + chunk_bad_crc = "FALSE" + if bad_crc and offset >= bad_crc_offset: + chunk_bad_crc = "TRUE" + bad_crc = None + + data = None + attempt = 0 + while data is None and attempt < MAX_ATTEMPTS: + attempt += 1 + # Record the received count so we know when the response arrives + count = tlm("INST2 MEMDUMP RECEIVED_COUNT") or 0 + + # Send the command, waiting for the link if it is down. This doesn't use up + # an attempt. An ack timeout may mean the command did reach the target but + # the duplicate response is harmless because it has the same OFFSET and SIZE. + link_deadline = time.time() + LINK_TIMEOUT + sent = False + while not sent: + try: + # The command CRC over OFFSET, SIZE and BAD_CRC is calculated automatically + cmd(f"INST2 MEMDUMP with OFFSET {offset}, SIZE {size}, BAD_CRC {chunk_bad_crc}") + sent = True + # Only the first request has a bad CRC so the retry succeeds + chunk_bad_crc = "FALSE" + except Exception as error: + if not re.search("not connected|waiting for cmd ack", str(error), re.IGNORECASE): + raise + if time.time() > link_deadline: + raise + print(f"WARN: {error}, retrying until the link returns") + wait(2) + + # Wait for the response to this request. A stale response (e.g. a late + # reply to an earlier request) is skipped and we keep waiting. + dump = None + deadline = time.time() + RESPONSE_TIMEOUT + while (remaining := deadline - time.time()) > 0: + # wait returns False on timeout rather than raising + if not wait(f"INST2 MEMDUMP RECEIVED_COUNT > {count}", remaining): + break + # Grab all the items at once so they all come from the same packet + response = {name: value for name, value, _ in get_tlm_packet("INST2 MEMDUMP", type="RAW")} + count = response["RECEIVED_COUNT"] + if response["OFFSET"] == offset and response["SIZE"] == size: + dump = response + break + print( + f"WARN: Ignoring stale response for OFFSET {response['OFFSET']}, SIZE {response['SIZE']} " + f"while waiting for OFFSET {offset}, SIZE {size}" + ) + if dump is None: + print(f"WARN: No response for OFFSET {offset} (attempt {attempt} of {MAX_ATTEMPTS})") + continue + + # Calculate the CRC in the script (the CRC_OK derived item does the same + # check on the ground for display on the screen) + chunk = bytes(dump["DATA"]) + calculated = crc.calc(chunk) + if calculated == dump["CRC"]: + print(f"CRC passed at OFFSET {offset}") + data = chunk + else: + mismatch = f"Target CRC 0x{dump['CRC']:08X} vs calculated 0x{calculated:08X}" + print(f"WARN: CRC mismatch at OFFSET {offset}: {mismatch}") + # Let the operator decide whether to request the chunk again + answer = message_box( + f"Bad CRC at OFFSET {offset}. Retry this chunk?", "Retry", "Stop Dump", informative=mismatch + ) + if answer == "Stop Dump": + raise RuntimeError(f"Dump stopped by operator at OFFSET {offset}") + # An operator retry doesn't use up an attempt + attempt -= 1 + if data is None: + raise RuntimeError(f"Unable to dump OFFSET {offset} after {MAX_ATTEMPTS} attempts") + dump_data += data + offset += size + +print(f"Dumped {len(dump_data)} bytes (CRC32 0x{crc.calc(dump_data):08X})") + +# Push the dump to the bucket +put_target_file(bucket_path, bytes(dump_data)) + +# Read it back to verify it made it to the bucket intact +file = get_target_file(bucket_path) +data = file.read() +file.close() +expected = bytes((START_OFFSET + i) & 0xFF for i in range(TOTAL_SIZE)) +check_expression(f"{len(data)} == {TOTAL_SIZE}") +check_expression(f"{crc.calc(data)} == {crc.calc(expected)}") +print(f"Memory dump saved to {bucket_path}") diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/streaming.py b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/streaming.py index 64aa9f4d38..b944cdb421 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/streaming.py +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/streaming.py @@ -1,4 +1,4 @@ -from datetime import datetime, timezone, timedelta +from datetime import UTC, datetime, timedelta from openc3.script.web_socket_api import StreamingWebSocketApi @@ -17,7 +17,7 @@ print(api.read()) # Warning this saves all data to RAM. Do not use for large queries -now = datetime.now(timezone.utc) +now = datetime.now(UTC) data = StreamingWebSocketApi.read_all( items=[ "DECOM__TLM__INST__HEALTH_STATUS__TEMP1__CONVERTED", diff --git a/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/screens/memdump.txt b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/screens/memdump.txt new file mode 100644 index 0000000000..960150c696 --- /dev/null +++ b/openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/screens/memdump.txt @@ -0,0 +1,40 @@ +SCREEN AUTO AUTO 1.0 + +TITLE "<%= target_name %> Memory Dump" + +HORIZONTAL + VERTICALBOX "Commands" + MATRIXBYCOLUMNS 2 + LABEL "Offset:" + NAMED_WIDGET OFFSET TEXTFIELD 10 "0" + LABEL "Size:" + NAMED_WIDGET SIZE TEXTFIELD 10 "256" + END + # Flips a random bit in the dump after the target calculates its CRC + NAMED_WIDGET BAD_CRC CHECKBUTTON 'Bad CRC' + # The command CRC over OFFSET, SIZE and BAD_CRC is calculated automatically + BUTTON 'Dump Memory' "var offset=screen.getNamedWidget('OFFSET').text();" + + "var size=screen.getNamedWidget('SIZE').text();" + + "var badCrc=screen.getNamedWidget('BAD_CRC').checked() ? 'TRUE' : 'FALSE';" + + "api.cmd('<%= target_name %> MEMDUMP with OFFSET '+offset+', SIZE '+size+', BAD_CRC '+badCrc)" + # cmd_raw skips the CRC write conversion so the target rejects the command + BUTTON 'Run Dump Script' "runScript('<%= target_name %>/procedures/memdump.py', true)" + BUTTON 'Run Dump Script with Bad CRC' "runScript('<%= target_name %>/procedures/memdump.py', true, {'BAD_CRC': '1'})" + END + + VERTICALBOX "Last Dump" + LABELVALUE <%= target_name %> MEMDUMP OFFSET + LABELVALUE <%= target_name %> MEMDUMP SIZE + LABELVALUE <%= target_name %> MEMDUMP CRC + LABELVALUE <%= target_name %> MEMDUMP CALC_CRC + LABELLED <%= target_name %> MEMDUMP CRC_OK + SETTING LED_COLOR TRUE GREEN + SETTING LED_COLOR FALSE RED + LABELVALUE <%= target_name %> MEMDUMP RECEIVED_COUNT + LABELVALUE <%= target_name %> MEMDUMP PACKET_TIME + END +END + +VERTICALBOX "Data" + BLOCK <%= target_name %> MEMDUMP DATA 680 200 "%02X" 4 8 "0x%08X: " +END diff --git a/openc3-cosmos-init/plugins/packages/openc3-js-common/src/services/cable.js b/openc3-cosmos-init/plugins/packages/openc3-js-common/src/services/cable.js index 4899bfb955..684bf54c1d 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-js-common/src/services/cable.js +++ b/openc3-cosmos-init/plugins/packages/openc3-js-common/src/services/cable.js @@ -23,6 +23,19 @@ import { isAuthRequiredError } from './authGuard' // enough that a tool isn't visibly stuck. const RECOVERY_DELAY = 1000 +const DEFAULT_URL = '/openc3-api/cable' + +// How often each subscription tells the cmd-tlm-api it is still alive. The +// server reaps broadcasters that stop heartbeating (see +// ApplicationCable::BroadcasterRegistry), which bounds the lifetime of a +// subscription whose disconnect never reached it. +const HEARTBEAT_INTERVAL = 60000 + +// Sent by the server in reply to a heartbeat when it no longer has a +// broadcaster for this subscription (e.g. it was reaped), asking us to +// resubscribe so the channel's subscribed callback runs again. +const RESUBSCRIBE_KEY = '__openc3_cable__' + // Wraps an anycable subscription so one that dies unrecoverably is replaced // transparently, keeping the object the caller holds valid across the swap. // @@ -57,6 +70,7 @@ class ResilientSubscription { this._additionalOptions = additionalOptions this._subscription = null this._unsubscribed = false + this._heartbeat = null } get identifier() { @@ -86,6 +100,7 @@ class ResilientSubscription { unsubscribe() { this._unsubscribed = true + this._stopHeartbeat() this._cable._forget(this) return this._subscription?.unsubscribe() } @@ -100,10 +115,20 @@ class ResilientSubscription { { channel: this._channel, token: localStorage.openc3Token, + // Tells the server this subscription heartbeats, so it can be reaped + // even if it dies before the first heartbeat is sent + ...(this._cable._heartbeatInterval ? { heartbeat: true } : {}), ...this._additionalOptions, }, { ...this._callbacks, + received: (data) => { + if (data?.[RESUBSCRIBE_KEY] === 'resubscribe') { + this._cable._scheduleRecovery() + return + } + this._callbacks.received?.(data) + }, disconnected: (data) => { // We tore this subscription down ourselves (unsubscribe, or the // whole cable disconnecting because the component unmounted). The @@ -122,13 +147,45 @@ class ResilientSubscription { }, }, ) + this._startHeartbeat() + } + + _startHeartbeat() { + const interval = this._cable._heartbeatInterval + if (!interval || this._heartbeat) { + return + } + this._heartbeat = setInterval(() => { + if (this._unsubscribed) { + this._stopHeartbeat() + return + } + // Failures (e.g. while disconnected) are expected and harmless; the next + // heartbeat or the reconnect's resubscribe covers them + Promise.resolve() + .then(() => this._subscription?.perform('heartbeat', {})) + .catch(() => {}) + }, interval) + } + + _stopHeartbeat() { + if (this._heartbeat) { + clearInterval(this._heartbeat) + this._heartbeat = null + } } } export default class Cable { - constructor(url = '/openc3-api/cable') { + // Heartbeats default on only for the cmd-tlm-api cable, whose channels + // track subscription liveness. Other backends may not accept the action. + constructor( + url = DEFAULT_URL, + { heartbeatInterval = url === DEFAULT_URL ? HEARTBEAT_INTERVAL : 0 } = {}, + ) { this._cable = null this._url = url + this._heartbeatInterval = heartbeatInterval this._subscriptions = new Set() this._recovery = null } @@ -140,6 +197,7 @@ export default class Cable { // recovery, since we just cancelled one and the consumer is going away. this._subscriptions.forEach((subscription) => { subscription._unsubscribed = true + subscription._stopHeartbeat() }) this._subscriptions.clear() if (this._cable) { diff --git a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue index 40f2e750ca..91b97fbc09 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue +++ b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue @@ -2048,6 +2048,8 @@ export default { items: items, start_time: theStartTime, end_time: this.graphEndDateTime, + // Ask to be told when realtime streaming skips data + notices: true, }) }) // An AuthRequiredError means we have no token and are being @@ -2141,6 +2143,11 @@ export default { }, received: function (data) { this.cable.recordPing() + // Data always arrives as an Array, notices we opted into are an Object + if (!Array.isArray(data)) { + this.receivedNotice(data) + return + } // Buffer incoming data and schedule processing via requestAnimationFrame // to prevent blocking the main thread during historical data floods this.pendingData.push(data) @@ -2150,6 +2157,47 @@ export default { ) } }, + receivedNotice: function (notice) { + // Ignore anything we don't understand so new notice types added on the + // server don't break older clients + if (notice.__type !== 'NOTICE' || notice.notice !== 'SKIPPED') { + return + } + // Coalesce repeated skips into one entry so a graph that can't keep up + // doesn't bury every other error, and only re-report it to an embedding + // screen (which keeps its own list) once a minute + const now = Date.now() + const existing = this.errors.find((error) => error.type === 'skipped') + // Accumulate the totals across every skip since the entry was created + const skippedSeconds = + (existing?.skippedSeconds || 0) + notice.skipped_seconds + const count = (existing?.count || 0) + 1 + let message = `Streaming fell behind realtime, skipped ${skippedSeconds.toFixed(1)}s of data` + if (count > 1) { + message += ` (${count} times)` + } + if (existing) { + // Update in place so the errors list shows the latest totals + existing.skippedSeconds = skippedSeconds + existing.count = count + existing.message = message + existing.time = now + // Throttle emits to the parent, which would otherwise add a new entry per skip + if (now - existing.emitted > 60000) { + existing.emitted = now + this.$emit('error', existing) + } + } else { + // First skip: addError stamps the time and emits to the parent + this.addError({ + type: 'skipped', + message: message, + skippedSeconds: skippedSeconds, + count: count, + emitted: now, + }) + } + }, // Drains the pendingData buffer in a time-boxed loop. Each call is // triggered by a single requestAnimationFrame scheduled in received(). // When processing exceeds MAX_PROCESSING_MS we schedule another frame diff --git a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/tools/admin/tabs/settings/AstroSettings.vue b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/tools/admin/tabs/settings/AstroSettings.vue index 0c284e07a9..7ec71ea2ae 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/tools/admin/tabs/settings/AstroSettings.vue +++ b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/tools/admin/tabs/settings/AstroSettings.vue @@ -30,12 +30,54 @@ color="primary" :disabled="settingsLoading" /> + +
+
+ Displays a second clock driven by a telemetry item such as + PACKET_TIMEFORMATTED or PACKET_TIMESECONDS. Formatted times are + treated as UTC and numeric values as seconds since the Unix epoch. +
+ + + + +
Save @@ -45,32 +87,84 @@