From 56aad3a0231d63ed458a717c27ce55bb863fb67c Mon Sep 17 00:00:00 2001 From: Alex-Emmet Date: Tue, 21 Jul 2026 23:05:34 +0200 Subject: [PATCH] [Fix] IPC, cli, daemon --- .forgejo/workflows/release.yml | 36 +- .gitignore | 5 + Cargo.lock | 11 +- config.json | 1 - dockerfile | 8 +- flake.nix | 69 ++- iota-cli/Cargo.toml | 3 +- iota-cli/src/elements/console_card.rs | 188 +------ iota-cli/src/ipc_client.rs | 590 ++++++++++++++++++-- iota-cli/src/screens/main_screen.rs | 80 ++- iota-core/Cargo.toml | 1 + iota-core/src/main.rs | 2 +- iota-daemon-lib/Cargo.toml | 4 + iota-daemon-lib/src/command_router.rs | 153 +++-- iota-daemon-lib/src/daemon_state.rs | 108 +++- iota-daemon-lib/src/ipc_server.rs | 191 ++++++- iota-daemon-lib/src/lib.rs | 2 +- iota-daemon/Cargo.toml | 1 + iota-daemon/src/main.rs | 84 ++- iota-ipc/src/lib.rs | 11 +- iota-ipc/src/protocol.rs | 124 +++- iota-ipc/src/transport.rs | 13 +- iota.mk | Bin 5713 -> 0 bytes iota/Cargo.toml | 1 + iota/src/main.rs | 4 +- omikron-connector/Cargo.toml | 1 + omikron-connector/src/omikron_connection.rs | 26 +- omikron-connector/src/user_ops.rs | 3 - systemd/iota-daemon.service | 12 +- systemd/iota-daemon.socket | 1 + web-server/Cargo.toml | 4 +- web-server/src/lib.rs | 18 +- 32 files changed, 1356 insertions(+), 399 deletions(-) delete mode 100644 config.json delete mode 100644 iota.mk diff --git a/.forgejo/workflows/release.yml b/.forgejo/workflows/release.yml index 06523ce..ec4e2af 100644 --- a/.forgejo/workflows/release.yml +++ b/.forgejo/workflows/release.yml @@ -37,16 +37,19 @@ jobs: DOCKER_PASSWD="$(printf '%s' "$DOCKER_PASSWD" | tr -d '\r\n')" printf '%s' "$DOCKER_PASSWD" | nix-shell -p docker --run "docker login docker.io --username \"$DOCKER_USER\" --password-stdin" - - name: Build & Push + - name: Build & Push Docker image run: | nix-shell -p docker --run "docker build -f dockerfile -t tensamin/iota:latest . && docker push tensamin/iota:latest" - - name: Build release binary + - name: Build release binaries run: | set -eu - nix build .#iota --print-build-logs - install -Dm755 result/bin/iota dist/iota + nix build .#iota-daemon --print-build-logs + install -Dm755 result/bin/iota-daemon dist/iota-daemon + + nix build .#iota-ui --print-build-logs + install -Dm755 result/bin/iota-ui dist/iota-ui - name: Read release metadata id: version @@ -55,7 +58,7 @@ jobs: run: | set -eu - VERSION="$(nix eval --raw .#iota.version)" + VERSION="$(nix eval --raw .#iota-daemon.version)" SHORT_SHA="$(git rev-parse --short=7 HEAD)" case "$RELEASE_TYPE" in @@ -73,18 +76,15 @@ jobs: ;; esac - ASSET_PATH="dist/iota" - ASSET_NAME="iota" - test -x "$ASSET_PATH" - echo "version=$VERSION" >> "$FORGEJO_OUTPUT" echo "tag=$TAG" >> "$FORGEJO_OUTPUT" echo "title=$TAG" >> "$FORGEJO_OUTPUT" echo "prerelease=$PRERELEASE" >> "$FORGEJO_OUTPUT" - echo "asset_path=$ASSET_PATH" >> "$FORGEJO_OUTPUT" - echo "asset_name=$ASSET_NAME" >> "$FORGEJO_OUTPUT" - - name: Create release and upload binary + test -x "dist/iota-daemon" + test -x "dist/iota-ui" + + - name: Create release and upload binaries env: TOKEN: ${{ forgejo.token }} API: ${{ forgejo.api_url }} @@ -93,8 +93,6 @@ jobs: TAG: ${{ steps.version.outputs.tag }} TITLE: ${{ steps.version.outputs.title }} PRERELEASE: ${{ steps.version.outputs.prerelease }} - ASSET_PATH: ${{ steps.version.outputs.asset_path }} - ASSET_NAME: ${{ steps.version.outputs.asset_name }} DESCRIPTION: ${{ inputs.description }} run: | nix-shell -p curl jq --run ' @@ -122,7 +120,11 @@ jobs: RELEASE_ID="$(echo "$RELEASE_JSON" | jq -r .id)" fi - curl -fsS -X POST "$API/repos/$REPO/releases/$RELEASE_ID/assets?name=$ASSET_NAME" \ + curl -fsS -X POST "$API/repos/$REPO/releases/$RELEASE_ID/assets?name=iota-daemon" \ -H "Authorization: token $TOKEN" \ - -F "attachment=@$ASSET_PATH" - ' \ No newline at end of file + -F "attachment=@dist/iota-daemon" + + curl -fsS -X POST "$API/repos/$REPO/releases/$RELEASE_ID/assets?name=iota-ui" \ + -H "Authorization: token $TOKEN" \ + -F "attachment=@dist/iota-ui" + ' diff --git a/.gitignore b/.gitignore index 257c89e..9fb3aaa 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,8 @@ target logs agreements languages/ +.envrc +.direnv +config.json +*.mk +*.sqlite* diff --git a/Cargo.lock b/Cargo.lock index 5329cde..96c66bb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2091,6 +2091,7 @@ version = "0.1.0" dependencies = [ "iota-cli", "tokio", + "tokio-util", ] [[package]] @@ -2186,6 +2187,7 @@ dependencies = [ "sysinfo", "tokio", "tokio-tungstenite", + "tokio-util", "tungstenite", "walkdir", "warp", @@ -2213,6 +2215,7 @@ dependencies = [ "iota-storage", "omikron-connector", "tokio", + "tokio-util", "web-server", ] @@ -2220,15 +2223,19 @@ dependencies = [ name = "iota-daemon-lib" version = "0.1.0" dependencies = [ + "dashmap", "iota-ipc", "iota-logger", "iota-state", "iota-storage", "iota-util", + "libc", "mtp", "omikron-connector", "sysinfo", "tokio", + "tokio-util", + "uuid", ] [[package]] @@ -3082,6 +3089,7 @@ dependencies = [ "reqwest", "sha2 0.10.9", "tokio", + "tokio-util", "uuid", "x448", ] @@ -4898,6 +4906,7 @@ dependencies = [ "bytes", "futures-core", "futures-sink", + "futures-util", "pin-project-lite", "tokio", ] @@ -5263,10 +5272,10 @@ dependencies = [ "bytes", "http 1.4.2", "iota-logger", - "iota-state", "iota-util", "mtp", "tokio", + "tokio-util", ] [[package]] diff --git a/config.json b/config.json deleted file mode 100644 index 03a370e..0000000 --- a/config.json +++ /dev/null @@ -1 +0,0 @@ -{"keyring":"BMAGjI76myen0MXtUh7jwKfZBzJNeBofMRcVcDTJVSZy5IXAiR+f9GrbpQdIdYMyEgSRlbr6oaiqIBkhaSJZkI6YaIUFyIn4aRvPhG//ET/1NwLLyLw/1ripyKgAzAEFC3lGaLn0U7AS93vPuTxxMm5koKR7hzG40rTjEkcu1w6j1lpMupmFKqwpyj4WC2koVybhln8G0MjO947yXE0Tk47GsnWr0SENM7b79AlXNZvL9AMsGQbdgYqjIbMRZWthKY4P6EiZoWKFuYJrixeaVWSr+mCACgRDaMT+os+w+jw56oNCpFbBuswFR3iF8I3p9JdbiTl0ya1HwrngZJlUub9KxRgfCz5ZRhjzqKZ7Kgwhy7S4Nx2EFH4UhEaf+Y3QrGHitn7YCZXB5U0YGT+cqYJe8br77KiLW5rvjG2RaoaoA4AF8jiJ2LjWo1UrZDWL+hmiISuyWy3jZE7njI6oqXTZMb5oMSnB5r4KCyy/nA4DEMiTcLIcRgyUAXx1sDGX2Qc4yXwXqoMxnDzlCBFrRwn3EFbLYRNGpUM4y8Az4AROtTwTJq714RU9mc9v2V7jW7uCxYCMWWG2N3RIcppCJRbuwy1nWEPxNGEs8lp0wgDaSU1exz4WBldSCLD/mcImlCNiC174FVBgfLjM7Hp8oKnhCSziaTmAQ8mm5300Spw+A66VvCvGxheosy9BuTCpBKe9lJ/XdLrfnHxud1CoYgh9ymtGXDfkZ2H1c70XyC70+US/mFhfK6osMMDqOsd5eIRF43E/0oQfUcL8CBxEFcbDRyumSbEZpz42hCb9ccCzJZf6x6Bd9mzUeZYBlibrHJ+TKA6I+quB9CD1dljIQjTxQyLC91vbayoPuQ9iF2DKoEnicCNhhzLzZHRqE2oIE3MirDsCmMGno0+Cs7aqF8mvC1jBuzIQwlotOW7CYLjORmWLOrbbGMfV+XtOogEDWVYrxSe3dUOZ2x3fcQzGiVXjGjv2eTYQgHafDJUs6hbtDCqKUYXjzILF7MkpGx0EGYGhEQBYkMfQ7GQ8fI1ppYbpVUYh6m6zIWnWXII29aZOGldKAwO6uSjBwDI9d7lN1JgVR35Iu4Wj5zJj6ro9QxJ/mnSTwINSdj4SRo7JKIgWgXkeoS5VGM4Tq4umnIiQPFPXObrXBhQH8Lezsqyq83qeQFydy699zMz/WACflUua8Hhlgk6lMV0VfB3yBoJQx1WRxhGQm44CeBgt18u/Wm+ZTFJ+IR/+4yUg+qRhuc74k83LsCorFSSfymE4UDY7XJbAuZgCFVTs6TjzVCS4+YMdC4Row2p83J+CsYUAYoCnV67GNKEtwxH7XAJZFyKkc3b/kEPffDYJ9Ep7yrEy5AEqaWaT7B8f6U/6Nw4feykfinaruFPGplduEY9P6sLQ4Fvn+wDtA5EMwbkXQLwJSkGv3GOHBWbIWY+HmWnUW6lGPJd9O8US88SFIG0e8BdW0I9IaiAuvCVgpw8ImLJDswndSGV4sJ0A0Un4qIQrzCcszAR+GKArhMXcGo1SSndo9MxROyg1uJFwfJM9BQMZJjAE2+7LaIyaAGIf+/zEGWBoFhPYQqXu13laMW/npdMRUR40YydPhBRuMkXMeiJaTDFYCYCLmHASQghali1XJXH2TMJOh59iC2RiuJeTAZD7IFy/gaQtOLCxoHHz6L0kODXCdJB0xkr8CETSkgf7fKKlxSC8+n31IqRCq1tsgqssahhRBRFy+aQ4NzhVypvdoKiFqo3XmbRPhl3QxGHweaw9PMcq55au2SuttA9kZUcRo0C9PAppO3kLkTSjNamLWBj36Lmuo2KNlQjm9AS2CCwixxOSIWBj+SNCiWl3TF/0uhb6NEiKk3hSHDYBmbT+mK7KbIMqWM/Yemw9SyOdlI3KiEGRuj2BWzoy+ARSGxyCZ4HxGryAaRtWo8AwSaKD+Dz97LcEYUTC5Ja4qE/2yLzNtVJimTiFpkJLaojVEieWyZOdks8smBwTqz9I1aiE4GqX0no1/C4aYxtauMSWGRh9KV/1dLnEcgRExTvf6GIzEH+/kEi4Rh7AJLGuvKZT9kwEm7Nb9Sw4oUmHlZhyh5Q/gGChQZ3cDDFLdi4TBkePuAk10Dz2u575m5GvwULlN0NaJ3hx46KvyWbaVw84pIYuNg1Dq1wXWx0dOM+djL7LyXrQvJaD+spC+b0+N0o0MYcmvHgsG0VYm4IORMR7qAZg3HiBzF8Sdp1+Fz6jJG/ffDJTSmPtYSQDTIwctXrtpmw2+wOVW3G/hsf8RS7kNVSFxUnzrKTjmjovCFOaO8KEigz/CpPre6iUiVHX+w9ZIMBX8zScuLoObArc9Bn3Ykt2IcCoJaPW94V6pcgCUlzGi5lWo4cAx7pcSlL0pTRkk5AbcYxq+74kNKm4FFwB+rfPnE1R2QxyWTbi1psFgHJGchjypQBGeU1fUg19RCfz5HAL5YCCuL+Z4a+QiIGbdBrgSzBgUI6Hg1oG+Kp7IqJehG5/ekuMZ3MtNo9f1MLB603zUmcV+CE9gliIGFtVpzCwNlyRIbICwpT/1kqEjHFiK2jh4AgaWJ8ZqWg/mb6/qBz2BsqEqEVm+AL8BQsBOjtxh8uD3Aoat7mSApxNgrKKSh0Y5Emv1woxhUjueSFu5zRBWayaYl9oSkbMGZTPMGAYyT3vVDn2PKWEV7t8ND27PBktUM6/8ifisKI/K8ZTqDG7N6qioz3iY6sCZ2+Fq6jUBjQwqx34qYfRmHXMsX6198Fv0rrG2WZmEWc6QTJG60S0Ez90oAf72LathCal5MMRICSWEsx8BJpugX0ccaEgxZ6yJ5N2wa4FEzvDZXxL1LAJY5A05TvEB7fNQUSS9TnSc0+tu88ZhMKYMQjpMjnncoH+iL/VIRrlC4rxeRshoxHQsDD/23GzWz+n9buPiFwkQ1T+eCc/Q0XRIgrqPFmlIizSfAPhaQcNdcmfRkIrdosFNXxg95ciKAD14mD6ynjRZwYLSC+YKgCXQm7g1mOu1aGUBJkcl1nlxrnVQFudlpq3+RPCgS8g6JCCxFHrkgoBeqWZS6KfoxiOdZm6uj8LIzah6bFn2BFKC4xTEDQRGwxCHBFF5DKqd69ExqTFcLWD53/d0HFLmBfv1V7ESnzinKK2mQGpqss0VwdVsy16tLv1EqgGjI76myen0MXtUh7jwKfZBzJNeBofMRcVcDTJVSZy5IXAiR+f9GrbpQdIdYMyEgSRlbr6oaiqIBkhaSJZkI6YaIUFyIn4aRvPhG//ET/1NwLLyLw/1ripyKgAzAEFC3lGaLn0U7AS93vPuTxxMm5koKR7hzG40rTjEkcu1w6j1lpMupmFKqwpyj4WC2koVybhln8G0MjO947yXE0Tk47GsnWr0SENM7b79AlXNZvL9AMsGQbdgYqjIbMRZWthKY4P6EiZoWKFuYJrixeaVWSr+mCACgRDaMT+os+w+jw56oNCpFbBuswFR3iF8I3p9JdbiTl0ya1HwrngZJlUub9KxRgfCz5ZRhjzqKZ7Kgwhy7S4Nx2EFH4UhEaf+Y3QrGHitn7YCZXB5U0YGT+cqYJe8br77KiLW5rvjG2RaoaoA4AF8jiJ2LjWo1UrZDWL+hmiISuyWy3jZE7njI6oqXTZMb5oMSnB5r4KCyy/nA4DEMiTcLIcRgyUAXx1sDGX2Qc4yXwXqoMxnDzlCBFrRwn3EFbLYRNGpUM4y8Az4AROtTwTJq714RU9mc9v2V7jW7uCxYCMWWG2N3RIcppCJRbuwy1nWEPxNGEs8lp0wgDaSU1exz4WBldSCLD/mcImlCNiC174FVBgfLjM7Hp8oKnhCSziaTmAQ8mm5300Spw+A66VvCvGxheosy9BuTCpBKe9lJ/XdLrfnHxud1CoYgh9ymtGXDfkZ2H1c70XyC70+US/mFhfK6osMMDqOsd5eIRF43E/0oQfUcL8CBxEFcbDRyumSbEZpz42hCb9ccCzJZf6x6Bd9mzUeZYBlibrHJ+TKA6I+quB9CD1dljIQjTxQyLC91vbayoPuQ9iF2DKoEnicCNhhzLzZHRqE2oIE3MirDsCmMGno0+Cs7aqF8mvC1jBuzIQwlotOW7CYLjORmWLOrbbGMfV+XtOogEDWVYrxSe3dUOZ2x3fcQzGiVXjGjv2eTYQgHafDJUs6hbtDCqKUYXjzILF7MkpGx0EGYGhEQBYkMfQ7GQ8fI1ppYbpVUYh6m6zIWnWXII29aZOGldKAwO6uSjBwDI9d7lN1JgVR35Iu4Wj5zJj6ro9QxJ/mnSTwINSdj4SRo7JKIgWgXkeoS5VGM4Tq4umnIiQPFPXObrXBhQH8Lezsqyq83qeQFydy699zMz/WACflUua8Hhlgk6lMV0VfB3yBoJQx1WRxhGQm44CeBgt18u/Wm+ZTFJ+IR/+4yUg+qRhuc74k83LsCorFSSfymE4UDY7XJbAuZgCFVTs6TjzVCS4+YMdC4Row2p83J+CsYUAYoCnV67GNKEtwxH7XAJZFyKkc3b/kEPffDYJ9Ep7yrEy5AEqaWaT7B8f6U/6Nw4feykfinaruFPGplduEY9P6sLQ4Fvn+wDtA5EMwbkXQLwJSkGv3GOHBWbIWY+HmWnUW6lGPJd9O8US88SFIG0e8BdW0I9IaiAuvCVgpw8ImLJDswndSGV4sJ0A0Un4qIQrzCcszAR+GKArhMXcGo1SSndo9MxROyg1uJFwfJM9BQMZJjAE2+7LaIyaAGIf+/zEGWBoFu+riy8UUa4SXlhFlkzX0LxFF3GjDM8Ig7Z2e5dQPjhTF1E2/gR7C+el3nZRt3rb4xZF9OMFK1MOEdxBoJorubl0sqA89NULY0ePfMRef3eqe4PsNwcCSlUyLzqWwM10uQegMln/lXI/TJfBe4wuQ0IrPW3GBv+60A1pKSEmgyeT9RLSupHZGxbZHz0C5wocLbcpelkae/caApr7vUp8Kt3npl89ZM3xVj6HSb/8UgolC3yXfwG/W9ssCzmtQ4vC6HRhzM29MGBjZxMeLbYY+jrbI23L/iIgnKVeCenTh3wIVcoEKfkRvs8sAj1EjicL/YMJ0InKoiOrFC4CjMcclcuMw30Y8FmUcXxZjqltqbw0lhyuGHJTcUdesostQBapvrDkXTRK6hViPJehixApUEeTTLSUJDu9Nw2/LmOTEkYn0+sVDmkyjyYvaPmH/pvzuDIizEnYkRR/PKopmXhtnuis2UEh9U57qkNfJc7aj/OL245w062GpHpd2ynssXfbq87mjo4kfXwWACVAuxf7RWlbDGTfEF2wYeY+EjaouU137mA66uWPFhE3gyd9pvr2BsljYmcn3h7YStAMVWPe2m1Rtqlcbaez+O0g8fdG2elJgMt3asOhOzgJPIi19wRORZjoPPqqlZiwS1mu2m4EX91hJK1ZUT/d9UckuhaFVSyKAFo4edEKS+AkGsB6OBPuuvwxzwgYCv11eXcX5evB+b0mpEB+KPj5wkjWQcK51w5/Ondv+YPzgv1QN5SJznlz3kTn22P+qT/Zohz8Esy8EHZclpQ+lg+aF78ybvq3A9LKqiABSIX6ojLlQ0zODOm8UeTg18y8j8ofn5ivy9z3D1ihX7wwkSenDkDK8Fi+9wKTconAfFBoH4LINDHwiQuUUQ6BgmD6xpB4df7UdKowuODZO7nOQRy4mQVyrJMlvf8Hm54zw4vURSI4T5kFhbBizqcWEVftE+dki4Qtxxtd0rbUeXsxfykdgDrfkiGs7IoJL+cFQrY4rea1650/NLvQqk5/BILe0JybYPkzJLlFqHrL/mX1nG1RZJMfq8ChH1WECs9cZehKRFg7umjCEBXsr7+P713HqtVti33HYlQIAJEawbsM5Fj56/nxQiKVabUAykQ5krDq5OhLtn/z2IQAexjHmBTmkSLVWGl8P7JY4BfFQeAL/mW5TxVxf1MR7UE/GY1Re214zkrX+aa5tdWq0QKRa1xm5iepHYKgbAeU3D2jOwVzfhMWs3vDe1o4HjQrgC2+065TpcY57kizDEr1sX1mDvpaZBpTMBvyVdwpQsdSSUK3lruHkvYW5ATV4sdhPUua2ATk/gAVHC2WkFcxxiesCqkhSlib6YgolYDzVJu21dtR1mxMJed8FADYB7ICL7JgFBpGcit0Da0Khs1DAfCq1VT1cgHvXymzsjQPUKCFsixNKbk7fzByrU2IYOokjPAi4vA9jwYMsQ6fedoM2uyQT2Ro4nJ1DhGH5DaGdsi3nwKs6ceOadRSc43uYoVEyzpFcQyHAqBMYnz5e6tSTCqynK5cd5EwzJl/ShHhfRNH3nWoFWnYYenD4SWDqvKmyvLuxq4gISAw3onDRJsYde9M6gGvG6lWZtH3g/94qqH2g6RugRM5n/+5riMiN5AYY5LCMgkHeM4eOur5HKymjw1EVGKnPsX2m00bSkWSrTwUiiv3ogjmzM+F52CU7+BjYhdnaGdW+3zDCxfL4LIC/10MJD4Btt5vsdZwnXQdDizb/OmuVkIp47fPDR2Ni8LlWB3HPLcSQiE2B9RBKSeDoC5rr02SvS6be9oB9gYigmEp3QkhxfNdszW0Fw7rIWmzRRumRid6zqJRK39gH1f6XHFIJkTluZPqMdbHG6ZFSMxevrpNwUsjO/R8FYUYLw6ntj4VhU54T3s6Zh/kYLKi+qiVwPkv7RchRxB1gsnUgGBhA5nuU1AHtoABZorru7Wfrc9PaYty6mPrbyNKE56OCkhL/U5xqcSVKaZuLij5Qc60KzN0ixumiCPmKWaCSwK5rrjdKjD0OhTz5gmFmmaEisuvh94o1N6tEe2vYFHVTQ0biDq8/AbmahaSLU1TAcCjIznvl77nLBG1MJdMIH4kzyIw7uhSz/m/aET88RxIvc399PuTlXyOuysCfPBig1AMiRnpBWPhIlh9KxFj53P8QtfZgfBrEU/bHe27bkxFWkwGUxJSKeqHcrt8H7NzqaQbw7vdGXB6TO5saqz8qX1oJUzF0Fiuq0ce2nLeC5tpGWNswQ/WHEJmyJPGUAGCYyWE88qi2AKmJ/XBgZnugSCmtHjDQ6iuHt3+a9ZWUEl0ZzEf03PgOHCu/fjr7yAlk62+s9CW5L6YWx2ZQlrbvKWik8N76JS5lerB+TYIPqy4uYw+1UVzVC2yjmuqdnBU1Mm922FQ+JdZ4tmHQpHdswXiQSwvZkLN3MAle2yRQbIgmBcAUaI+rFBw4wUYM3UtAjjQoyKA2ATtRs7539E7KjjgNmpeHY/8ezDH+VqHscdOp1vncoMWI7AGgdai9LRnsGttSMddGMr2tgizkL6ucvIX5P8Oi57actaPRjeMRIsSyop9HQblrQhffJ1y0GlKwxI1gZQCqf955zn5P1uGCHGSh07qvP0e1qySPOb6rGkNhRV5xpPQ+Ik9YJBtx23j6kGwRI0DBFucoHJzRXsdgZZdn7MJ//jztsNzgVpx65QT8AAEeEX4qHHR8AVDqq49MrCQihEAIFXhH997tCM+mQuv4UGlq0pcTU3lxXZU/iFXBcKGNM57ACCjaD7MAxirp1DcoqFvmBAoogV577OS9UCCo+4cgL0MewAgXmP0NeyVgq5Dg60GcRXJ5P+4ulE/8wQX5boc1bhFEYY="} \ No newline at end of file diff --git a/dockerfile b/dockerfile index 5f07197..fcbfafa 100644 --- a/dockerfile +++ b/dockerfile @@ -4,7 +4,7 @@ FROM rust:latest AS builder WORKDIR /app COPY . . -RUN cargo build --release +RUN cargo build --release -p iota-daemon # Runtime stage FROM debian:sid @@ -13,8 +13,10 @@ WORKDIR /app RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/* -COPY --from=builder /app/target/release/iota-core . +COPY --from=builder /app/target/release/iota-daemon . + +RUN useradd -r -s /bin/false iota && mkdir -p /run/iota && chown iota:iota /run/iota EXPOSE 1984 -CMD ["./iota-core"] +CMD ["./iota-daemon"] diff --git a/flake.nix b/flake.nix index 607fb77..f811ec9 100644 --- a/flake.nix +++ b/flake.nix @@ -38,10 +38,13 @@ rustToolchain = rustPkgs.rust-bin.stable.latest.default.override { extensions = ["rust-src" "rust-analyzer" "clippy" "rustfmt"]; }; + commonBuildInputs = with pkgs; [openssl sqlite]; + commonNativeBuildInputs = with pkgs; [cmake perl pkg-config]; in { packages = { - default = self'.packages.iota; - iota = pkgs.rustPlatform.buildRustPackage { + default = self'.packages.iota-daemon; + + iota-daemon = pkgs.rustPlatform.buildRustPackage { pname = "iota-daemon"; version = "0.1.0"; src = ./.; @@ -50,8 +53,8 @@ lockFile = ./Cargo.lock; allowBuiltinFetchGit = true; }; - nativeBuildInputs = with pkgs; [cmake perl pkg-config]; - buildInputs = with pkgs; [openssl sqlite]; + nativeBuildInputs = commonNativeBuildInputs; + buildInputs = commonBuildInputs; dontUseCmakeConfigure = true; postInstall = '' for f in $out/bin/*; do @@ -62,11 +65,36 @@ ''; passthru.dataDir = "/var/lib/iota"; }; + + iota-ui = pkgs.rustPlatform.buildRustPackage { + pname = "iota-ui"; + version = "0.1.0"; + src = ./.; + cargoBuildFlags = ["-p" "iota"]; + cargoLock = { + lockFile = ./Cargo.lock; + allowBuiltinFetchGit = true; + }; + nativeBuildInputs = commonNativeBuildInputs; + buildInputs = commonBuildInputs; + dontUseCmakeConfigure = true; + postInstall = '' + for f in $out/bin/*; do + if [ "$(basename "$f")" != "iota" ]; then + rm "$f" + fi + done + # Rename to avoid confusion + if [ -f "$out/bin/iota" ]; then + mv "$out/bin/iota" "$out/bin/iota-ui" + fi + ''; + }; }; devShells.default = pkgs.mkShell { nativeBuildInputs = with pkgs; [rustToolchain git cmake perl pkg-config]; - buildInputs = with pkgs; [openssl sqlite]; + buildInputs = commonBuildInputs; }; }; @@ -78,7 +106,7 @@ ... }: let cfg = config.services.iota; - defaultPackage = self.packages.${pkgs.stdenv.hostPlatform.system}.default or (throw "iota: no pre-built package for system ${pkgs.stdenv.hostPlatform.system}"); + defaultPackage = self.packages.${pkgs.stdenv.hostPlatform.system}.iota-daemon or (throw "iota: no pre-built package for system ${pkgs.stdenv.hostPlatform.system}"); configFile = if cfg.settingsFile != null @@ -158,14 +186,27 @@ users.groups.iota = {}; - systemd.services.iota = { + systemd.sockets.iota-daemon = { + description = "${descriptionText} IPC socket"; + wantedBy = ["sockets.target"]; + socketConfig = { + ListenStream = "/run/iota/iota.sock"; + SocketMode = "0660"; + SocketUser = "iota"; + SocketGroup = "iota"; + Backlog = 5; + RemoveOnStop = "true"; + }; + }; + + systemd.services.iota-daemon = { description = descriptionText; - wantedBy = ["multi-user.target"]; after = ["network.target"]; + requires = ["iota-daemon.socket"]; serviceConfig = { - Type = "simple"; + Type = "notify"; User = "iota"; Group = "iota"; WorkingDirectory = cfg.dataDir; @@ -186,11 +227,19 @@ '') ]; - Restart = "always"; + Restart = "on-failure"; RestartSec = "5s"; RuntimeDirectory = "iota"; RuntimeDirectoryMode = "0750"; + # Exit code 75 = restart requested + RestartPreventExitStatus = "0"; + RestartForceExitStatus = "75"; + + TimeoutStopSec = "10"; + KillMode = "mixed"; + KillSignal = "SIGTERM"; + AmbientCapabilities = ["CAP_NET_BIND_SERVICE"]; CapabilityBoundingSet = ["CAP_NET_BIND_SERVICE"]; diff --git a/iota-cli/Cargo.toml b/iota-cli/Cargo.toml index 2c5f868..f79241a 100644 --- a/iota-cli/Cargo.toml +++ b/iota-cli/Cargo.toml @@ -17,7 +17,7 @@ iota-logger = { path = "../iota-logger", optional = true } iota-state = { path = "../iota-state" } iota-storage = { path = "../iota-storage", optional = true } iota-terms = { path = "../iota-terms" } - iota-util = { path = "../iota-util", optional = true } +iota-util = { path = "../iota-util", optional = true } iota-ipc = { path = "../iota-ipc" } omikron-connector = { path = "../omikron-connector", optional = true } @@ -64,6 +64,7 @@ strum = "0.27.2" strum_macros = "0.27.2" sysinfo = "0.38.3" tokio = { version = "1.50.0", features = ["full"] } +tokio-util = { version = "0.7", features = ["rt"] } tokio-tungstenite = { version = "*", features = ["native-tls"] } tungstenite = "*" walkdir = "2.5.0" diff --git a/iota-cli/src/elements/console_card.rs b/iota-cli/src/elements/console_card.rs index 5ece1d9..276b95f 100644 --- a/iota-cli/src/elements/console_card.rs +++ b/iota-cli/src/elements/console_card.rs @@ -1,18 +1,4 @@ use crossterm::event::{KeyCode, KeyEvent}; -#[cfg(feature = "legacy-commands")] -use iota_logger::{log, log_cv}; -#[cfg(feature = "legacy-commands")] -use iota_state::{ACTIVE_TASKS, RELOAD, SHUTDOWN}; -#[cfg(feature = "legacy-commands")] -use iota_storage::users::{user_manager, user_profile::UserProfile}; -#[cfg(feature = "legacy-commands")] -use iota_storage::util::config_util::modify_config; -#[cfg(feature = "legacy-commands")] -use iota_util::file_util; -#[cfg(feature = "legacy-commands")] -use mtp::codec::{CommunicationType, CommunicationValue}; -#[cfg(feature = "legacy-commands")] -use omikron_connector::omikron_connection::OMIKRON_CONNECTION; use ratatui::{ Frame, layout::Rect, @@ -47,7 +33,7 @@ pub struct ConsoleCard { cursor: Arc>, last_swap: Arc>, - tab_index: usize, + pending_restore: Arc>>, } impl ConsoleCard { @@ -62,7 +48,7 @@ impl ConsoleCard { joins: Borders::NONE, cursor: Arc::new(Mutex::new(true)), last_swap: Arc::new(Mutex::new(Instant::now())), - tab_index: 0, + pending_restore: Arc::new(Mutex::new(None)), } } @@ -112,12 +98,12 @@ impl ConsoleCard { spans.push(Span::styled(" ", Style::default().fg(Color::White))); } spans.push(Span::styled( - "send command ( for info)", + "send command (/help for info)", Style::default().fg(Color::DarkGray), )); } else { spans.push(Span::styled( - " send command ( for info)", + " send command (/help for info)", Style::default().fg(Color::DarkGray), )); } @@ -295,6 +281,12 @@ impl InteractableElement for ConsoleCard { } fn interact(&mut self, key: KeyEvent) -> InteractionResult { + // Check if a previously failed command should be restored. + if let Some(restored) = self.pending_restore.lock().unwrap().take() { + self.content = restored; + self.cursor_position = self.content.chars().count(); + } + match key.code { KeyCode::Enter => { if self.content.is_empty() { @@ -303,17 +295,15 @@ impl InteractableElement for ConsoleCard { let command = self.content.clone(); let ipc = self.ipc.clone(); - let seq = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as u64; + let restore = self.pending_restore.clone(); tokio::spawn(async move { - let _ = ipc.send_command(seq, command).await; + if ipc.send_command(0, command.clone()).await.is_err() { + *restore.lock().unwrap() = Some(command); + } }); self.content.clear(); self.cursor_position = 0; - self.tab_index = 0; InteractionResult::Handled } KeyCode::Backspace => { @@ -350,14 +340,7 @@ impl InteractableElement for ConsoleCard { self.cursor_position = self.content.chars().count(); InteractionResult::Handled } - KeyCode::Tab => { - if let Some(prefix) = self.current_prefix() { - if prefix == "/" { - self.tab_index = self.tab_index.saturating_add(1); - } - } - InteractionResult::Handled - } + KeyCode::Tab | KeyCode::BackTab => InteractionResult::Unhandled, _ => { if let Some(c) = key.code.as_char() { self.insert_at_cursor(c); @@ -381,144 +364,3 @@ impl InteractableElement for ConsoleCard { self.focused = f; } } - -#[cfg(feature = "legacy-commands")] -pub async fn run_command(command: &str) { - let parts = command.split(" ").collect::>(); - - match parts.as_slice() { - ["tasks"] => { - let active_tasks: Vec = - ACTIVE_TASKS.clone().iter().map(|v| v.to_string()).collect(); - let info = if *SHUTDOWN.read().await && *RELOAD.read().await { - "Rebooting, " - } else if *SHUTDOWN.read().await { - "Shutting , " - } else { - "" - }; - log!("{}Active tasks: {:?}", info, active_tasks); - } - ["fps"] => { - let (fps, skips) = *FPS.read().await; - log!("{:.1} FPS with {:.1}% of attempts skipped", fps, skips); - } - - ["help"] => { - log!("Available commands: tasks, fps, ping, user, reconnect, regenerate"); - } - - ["help", "tasks"] => { - log!("Tasks command usage: tasks"); - } - ["help", "fps"] => { - log!("FPS command usage: fps"); - } - ["help", "ping"] => { - log!("Ping command usage: ping [time]"); - } - ["help", "user"] => { - log!("User command usage: user add | user remove | user list"); - } - ["help", "reconnect"] => { - log!("Reconnect command usage: reconnect. Retry connecting to the Omikron server"); - } - ["help", "regenerate"] => { - log!( - "Regenerate command usage: regenerate keys. Generate a new Iota key pair and reconnect" - ); - } - - ["ping"] => { - ping(20).await; - } - ["ping", time] => { - let time = time.parse::().unwrap_or(20); - ping(time).await; - } - ["user", "add", username] => { - if let (Some(user), Some(_)) = omikron_connector::user_ops::create_user(username).await - { - log!("Created user {}", user.user_id); - } else { - log!("User creation: Failed to create user. See errors above."); - } - } - ["user", "remove", username] => { - if let Some(user) = user_manager::get_user_by_username(username) { - let msg = CommunicationValue::new(CommunicationType::DeleteUser) - .with_sender(user.user_id as u64); - let _ = OMIKRON_CONNECTION.send_message(&msg).await; - user_manager::remove_user(user.user_id); - log!("Removed user {}", user.user_id); - } else { - log!("User removal: Username doesn't exist"); - } - } - ["user", "list"] => { - let users: Vec = user_manager::get_users(); - for user in users { - let storage = file_util::get_designed_storage(user.user_id); - log!( - "> Username: {}, ID: {}, created at: {}, storage: {}", - user.username, - user.user_id, - user.created_at, - storage - ); - } - } - ["user", "info", username] => { - if let Some(user) = user_manager::get_user_by_username(username) { - user_manager::remove_user(user.user_id); - log!("Removed user {}", user.user_id); - } else { - log!("User info: Username doesn't exist"); - } - } - ["reconnect"] => { - log!("Reconnecting to Omikron server..."); - OMIKRON_CONNECTION.reconnect().await; - log!("Reconnected to Omikron server"); - } - ["regenerate", "keys"] => { - log!("Regenerating Iota key pair..."); - modify_config(|cfg| { - cfg.public_key = None; - cfg.private_key = None; - cfg.iota_id = None; - }); - log!("Key pair regenerated. Reconnecting to Omikron server..."); - OMIKRON_CONNECTION.reconnect().await; - log!("Reconnected with new key pair"); - } - ["reload"] | ["restart"] => { - log!("Restarting"); - *RELOAD.write().await = true; - *SHUTDOWN.write().await = true; - } - ["shutdown"] | ["stop"] => { - log!("Shutting down"); - *SHUTDOWN.write().await = true; - } - _ => { - log!("Unknown command"); - } - } -} - -#[cfg(feature = "legacy-commands")] -pub async fn ping(time: u64) { - let conn = OMIKRON_CONNECTION.clone(); - - let response_cv = conn - .await_response( - &CommunicationValue::new(CommunicationType::Ping), - Some(Duration::from_secs(time)), - ) - .await; - match response_cv { - Ok(response) => log_cv!(response), - Err(err) => log!("Ping error: {:?}", err), - } -} diff --git a/iota-cli/src/ipc_client.rs b/iota-cli/src/ipc_client.rs index c594f7e..018cfa2 100644 --- a/iota-cli/src/ipc_client.rs +++ b/iota-cli/src/ipc_client.rs @@ -1,43 +1,472 @@ -use iota_ipc::{ClientMessage, DaemonMessage, read_msg, write_msg}; +use iota_ipc::{ + ClientMessage, DaemonMessage, LocalRequest, MIN_PROTOCOL_VERSION, PROTOCOL_VERSION, + RequestEnvelope, ResponseResult, read_msg, write_msg, +}; use iota_state::{ClientState, UiLogEntry}; +use std::collections::HashMap; use std::io::Result; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; use tokio::net::UnixStream; -use tokio::net::unix::OwnedWriteHalf; -use tokio::sync::Mutex; +use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::sync::{Mutex, oneshot, watch}; + +const INITIAL_BACKOFF: Duration = Duration::from_millis(200); +const MAX_BACKOFF: Duration = Duration::from_secs(10); +const MAX_RECONNECT_ATTEMPTS: u32 = 50; + +/// Connection state exposed to the UI. +#[derive(Clone, Debug)] +pub enum IpcConnectionState { + Connecting, + Connected, + Reconnecting { attempt: u32 }, + Incompatible { message: String }, + Disconnected, +} + +/// Pending request awaiting a response. +struct PendingRequest { + response_tx: oneshot::Sender, +} /* The TUI owns this cache. IPC updates replace daemon snapshots and append * logs, so rendering never reaches into daemon-owned storage or connections. */ pub struct IpcClient { state: ClientState, writer: Mutex, + next_request_id: AtomicU64, + pending: Mutex>, + connection_state: watch::Sender, + path: PathBuf, } impl IpcClient { pub async fn connect(path: impl AsRef) -> Result> { - let stream = UnixStream::connect(path).await?; + let path = path.as_ref().to_path_buf(); + let stream = Self::try_connect(&path).await?; let (mut reader, writer) = stream.into_split(); + + let (conn_state_tx, _) = watch::channel(IpcConnectionState::Connecting); let client = Arc::new(Self { state: ClientState::new(), writer: Mutex::new(writer), + next_request_id: AtomicU64::new(1), + pending: Mutex::new(HashMap::new()), + connection_state: conn_state_tx, + path: path.clone(), }); + + // --- Handshake: send Hello, read HelloAck --- + { + let mut w = client.writer.lock().await; + write_msg( + &mut *w, + &ClientMessage::Hello { + supported_versions: vec![MIN_PROTOCOL_VERSION, PROTOCOL_VERSION], + }, + ) + .await?; + } + match read_msg::<_, DaemonMessage>(&mut reader).await { + Ok(DaemonMessage::HelloAck(ack)) => { + if ack.protocol_version < MIN_PROTOCOL_VERSION { + let _ = client + .connection_state + .send(IpcConnectionState::Incompatible { + message: format!( + "Daemon protocol {} < required {}", + ack.protocol_version, MIN_PROTOCOL_VERSION + ), + }); + return Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + format!( + "Protocol version mismatch: daemon={}, minimum={}", + ack.protocol_version, MIN_PROTOCOL_VERSION + ), + )); + } + } + Ok(_) => { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Expected HelloAck from daemon", + )); + } + Err(e) => return Err(e), + } + + let _ = client.connection_state.send(IpcConnectionState::Connected); + + // Start reader task (continues reading after handshake) let reader_client = client.clone(); tokio::spawn(async move { - while let Ok(message) = read_msg::<_, DaemonMessage>(&mut reader).await { - reader_client.apply(message).await; - } + reader_client.read_loop(reader).await; }); - client.send(ClientMessage::Subscribe).await?; + + // Subscribe to events + client + .send(ClientMessage::Subscribe { + log_classes: vec![], + metric_interval_ms: Some(500), + }) + .await?; + Ok(client) } + /// Try to connect with retries for socket activation. + pub async fn connect_or_activate(path: impl AsRef) -> Result> { + let path = path.as_ref().to_path_buf(); + let max_attempts = 30; + for attempt in 0..max_attempts { + match Self::connect(&path).await { + Ok(client) => return Ok(client), + Err(error) => { + if attempt < max_attempts - 1 { + let delay = Duration::from_millis(100 + attempt as u64 * 100); + tokio::time::sleep(delay).await; + continue; + } + return Err(error); + } + } + } + unreachable!() + } + + async fn try_connect(path: &Path) -> Result { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + match UnixStream::connect(path).await { + Ok(stream) => return Ok(stream), + Err(error) => { + if tokio::time::Instant::now() >= deadline { + return Err(error); + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + } + } + + /// Start the reconnection actor. + pub fn spawn_reconnector(self: &Arc) { + let client = self.clone(); + tokio::spawn(async move { + client.reconnection_loop().await; + }); + } + + async fn reconnection_loop(self: Arc) { + let mut rx = self.connection_status(); + + loop { + // Wait until the connection enters the Disconnected state. + loop { + let disconnected = matches!(*rx.borrow(), IpcConnectionState::Disconnected); + if disconnected { + break; + } + if rx.changed().await.is_err() { + return; // sender dropped + } + } + + let mut backoff = INITIAL_BACKOFF; + let mut attempt: u32 = 0; + + // Attempt reconnection until success or max attempts. + loop { + tokio::time::sleep(backoff).await; + attempt += 1; + + if attempt > MAX_RECONNECT_ATTEMPTS { + let _ = self + .connection_state + .send(IpcConnectionState::Incompatible { + message: "Max reconnection attempts exceeded".into(), + }); + return; + } + + let _ = self + .connection_state + .send(IpcConnectionState::Reconnecting { attempt }); + + match Self::try_connect(&self.path).await { + Ok(stream) => { + let (mut reader, writer) = stream.into_split(); + *self.writer.lock().await = writer; + + // Re-handshake + { + let mut w = self.writer.lock().await; + if write_msg( + &mut *w, + &ClientMessage::Hello { + supported_versions: vec![ + MIN_PROTOCOL_VERSION, + PROTOCOL_VERSION, + ], + }, + ) + .await + .is_err() + { + let _ = self + .connection_state + .send(IpcConnectionState::Disconnected); + break; + } + } + + // Read HelloAck + match read_msg::<_, DaemonMessage>(&mut reader).await { + Ok(DaemonMessage::HelloAck(ack)) => { + if ack.protocol_version < MIN_PROTOCOL_VERSION { + let _ = self.connection_state.send( + IpcConnectionState::Incompatible { + message: format!( + "Daemon protocol {} < required {}", + ack.protocol_version, MIN_PROTOCOL_VERSION + ), + }, + ); + return; + } + } + _ => { + let _ = self + .connection_state + .send(IpcConnectionState::Disconnected); + break; + } + } + + // Clear pending requests with connection-lost errors + { + let mut pending = self.pending.lock().await; + for (_, request) in pending.drain() { + let _ = request.response_tx.send( + ResponseResult::Error( + iota_ipc::IpcErrorCode::Disconnected, + ), + ); + } + } + + let _ = self.connection_state.send(IpcConnectionState::Connected); + + // Start new reader loop + let reader_client = self.clone(); + tokio::spawn(async move { + reader_client.read_loop(reader).await; + }); + + // Resubscribe + let _ = self + .send(ClientMessage::Subscribe { + log_classes: vec![], + metric_interval_ms: Some(500), + }) + .await; + + // Successfully reconnected; go back to waiting for + // the next disconnect. + break; + } + Err(_) => { + backoff = std::cmp::min(backoff * 2, MAX_BACKOFF); + } + } + } + } + } + + async fn read_loop(self: Arc, mut reader: OwnedReadHalf) { + loop { + match read_msg::<_, DaemonMessage>(&mut reader).await { + Ok(message) => self.apply(message).await, + Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => { + let _ = self.connection_state.send(IpcConnectionState::Disconnected); + break; + } + Err(_) => { + let _ = self.connection_state.send(IpcConnectionState::Disconnected); + break; + } + } + } + } + pub fn state(&self) -> ClientState { self.state.clone() } - pub async fn send_command(&self, seq: u64, line: String) -> Result<()> { - self.send(ClientMessage::Command { seq, line }).await + pub fn connection_status(&self) -> watch::Receiver { + self.connection_state.subscribe() + } + + pub fn connection_status_snapshot(&self) -> IpcConnectionState { + self.connection_state.borrow().clone() + } + + pub async fn send_request(&self, request: LocalRequest) -> Result { + let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed); + let (response_tx, response_rx) = oneshot::channel(); + + { + let mut pending = self.pending.lock().await; + pending.insert(request_id, PendingRequest { response_tx }); + } + + let envelope = RequestEnvelope { + request_id, + protocol_version: PROTOCOL_VERSION, + request, + }; + self.send(ClientMessage::Request(envelope)).await?; + + match tokio::time::timeout(Duration::from_secs(30), response_rx).await { + Ok(Ok(result)) => Ok(result), + Ok(Err(_)) => Ok(ResponseResult::Error(iota_ipc::IpcErrorCode::Disconnected)), + Err(_) => { + self.pending.lock().await.remove(&request_id); + Ok(ResponseResult::Error(iota_ipc::IpcErrorCode::Disconnected)) + } + } + } + + /// Parse a legacy console command string into a typed request. + pub fn parse_console_command(line: &str) -> Option { + let parts: Vec<&str> = line.trim_start_matches('/').split_whitespace().collect(); + match parts.as_slice() { + ["help"] => None, + ["tasks"] => Some(LocalRequest::ListTasks), + ["user", "add", username] => Some(LocalRequest::CreateUser { + username: username.to_string(), + }), + ["user", "remove", user_id_str] => { + let user_id = user_id_str.parse::().ok()?; + Some(LocalRequest::RemoveUser { user_id }) + } + ["user", "list"] => Some(LocalRequest::ListUsers), + ["reconnect"] => Some(LocalRequest::ReconnectOmikron), + ["regenerate", "keys"] => Some(LocalRequest::RotateIotaIdentity), + ["reload"] | ["restart"] => Some(LocalRequest::RestartDaemon), + ["shutdown"] | ["stop"] => Some(LocalRequest::StopDaemon), + _ => None, + } + } + + /// Legacy command interface: parse text command, send as typed request. + pub async fn send_command(&self, _seq: u64, line: String) -> Result<()> { + let trimmed = line.trim_start_matches('/').trim(); + + // Handle ping as a direct Ping message (not a LocalRequest). + if trimmed == "ping" || trimmed.starts_with("ping ") { + let seq = self.next_request_id.fetch_add(1, Ordering::Relaxed); + if let Err(e) = self.send(ClientMessage::Ping { seq }).await { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + state.push_log(UiLogEntry { + timestamp_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + sender: "Console".into(), + message: format!("Failed to send ping: {}", e), + is_error: true, + }); + return Err(e); + } + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + state.push_log(UiLogEntry { + timestamp_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + sender: "Console".into(), + message: "Ping sent".into(), + is_error: false, + }); + return Ok(()); + } + + if let Some(request) = Self::parse_console_command(&line) { + match self.send_request(request).await { + Ok(result) => { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + let message = match &result { + ResponseResult::Ok(msg) => msg.clone(), + ResponseResult::Error(code) => format!("Error: {:?}", code), + }; + state.push_log(UiLogEntry { + timestamp_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + sender: "Command".into(), + message, + is_error: matches!(&result, ResponseResult::Error(_)), + }); + Ok(()) + } + Err(e) => { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + state.push_log(UiLogEntry { + timestamp_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + sender: "Console".into(), + message: format!("Failed to send: {}", e), + is_error: true, + }); + Err(e) + } + } + } else { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + state.push_log(UiLogEntry { + timestamp_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + sender: "Console".into(), + message: if line.trim() == "help" { + "Available commands: tasks, ping, user, reconnect, regenerate, restart, stop" + .into() + } else { + format!("Unknown command: {}", line) + }, + is_error: false, + }); + Ok(()) + } } async fn send(&self, message: ClientMessage) -> Result<()> { @@ -46,19 +475,26 @@ impl IpcClient { } async fn apply(&self, message: DaemonMessage) { - let mut state = self - .state - .app - .lock() - .unwrap_or_else(|error| error.into_inner()); match message { - DaemonMessage::LogEntry(entry) => state.push_log(UiLogEntry { - timestamp_ms: entry.timestamp_ms, - sender: entry.sender, - message: entry.message, - is_error: entry.is_error, - }), + DaemonMessage::LogEntry(entry) => { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + state.push_log(UiLogEntry { + timestamp_ms: entry.timestamp_ms, + sender: entry.sender, + message: entry.message, + is_error: entry.is_error, + }); + } DaemonMessage::StateUpdate(snapshot) => { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); state.cpu = snapshot.cpu; state.ram = snapshot.ram; state.ping = snapshot.ping; @@ -66,18 +502,106 @@ impl IpcClient { state.net_down = snapshot.net_down; state.sys_info = snapshot.sys_info; } - DaemonMessage::CommandResult { - success, message, .. - } => state.push_log(UiLogEntry { - timestamp_ms: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis(), - sender: "Command".into(), - message, - is_error: !success, - }), + DaemonMessage::MetricSample(sample) => { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + if let Some(cpu) = sample.cpu { + let idx = state.cpu.len() as f64; + state.cpu.push((idx, cpu)); + if state.cpu.len() > iota_state::MAX_POINTS { + state.cpu.remove(0); + } + } + if let Some(ram) = sample.ram { + let idx = state.ram.len() as f64; + state.ram.push((idx, ram)); + if state.ram.len() > iota_state::MAX_POINTS { + state.ram.remove(0); + } + } + if let Some(ping) = sample.ping { + state.push_ping_val(ping); + } + if let Some(net_up) = sample.net_up { + let idx = state.net_up.len() as f64; + state.net_up.push((idx, net_up)); + if state.net_up.len() > iota_state::MAX_POINTS { + state.net_up.remove(0); + } + } + if let Some(net_down) = sample.net_down { + let idx = state.net_down.len() as f64; + state.net_down.push((idx, net_down)); + if state.net_down.len() > iota_state::MAX_POINTS { + state.net_down.remove(0); + } + } + } + DaemonMessage::Response(response) => { + let mut pending = self.pending.lock().await; + if let Some(request) = pending.remove(&response.request_id) { + let _ = request.response_tx.send(response.result); + } else { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + let message = match &response.result { + ResponseResult::Ok(msg) => msg.clone(), + ResponseResult::Error(code) => format!("Error: {:?}", code), + }; + state.push_log(UiLogEntry { + timestamp_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + sender: "Command".into(), + message, + is_error: matches!(&response.result, ResponseResult::Error(_)), + }); + } + } + DaemonMessage::HelloAck(_) => {} DaemonMessage::Pong { .. } => {} + DaemonMessage::LifecycleEvent(event) => match event { + iota_ipc::LifecycleEvent::Shutdown { reason } => { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + state.push_log(UiLogEntry { + timestamp_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + sender: "Daemon".into(), + message: format!("Daemon shutting down: {}", reason), + is_error: true, + }); + } + _ => {} + }, + DaemonMessage::Gap { skipped } => { + let mut state = self + .state + .app + .lock() + .unwrap_or_else(|error| error.into_inner()); + state.push_log(UiLogEntry { + timestamp_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(), + sender: "System".into(), + message: format!("Skipped {} messages, resynchronizing", skipped), + is_error: false, + }); + } } } } diff --git a/iota-cli/src/screens/main_screen.rs b/iota-cli/src/screens/main_screen.rs index 576afeb..49828c2 100644 --- a/iota-cli/src/screens/main_screen.rs +++ b/iota-cli/src/screens/main_screen.rs @@ -6,6 +6,7 @@ use crate::{ log_card::LogCard, }, interaction_result::InteractionResult, + ipc_client::IpcConnectionState, screens::screens::{NavDirection, Screen}, ui::UI, }; @@ -16,6 +17,7 @@ use ratatui::{ layout::{Constraint, Layout, Margin, Rect}, widgets::{Block, Borders}, }; +use tokio::sync::watch; use std::{any::Any, sync::Arc}; @@ -24,6 +26,7 @@ pub struct MainScreen { nav_grid: Vec>>, selected_coords: (usize, usize), graphs_open: bool, + connection_status_rx: watch::Receiver, } impl MainScreen { @@ -58,11 +61,14 @@ impl MainScreen { let graphs_open = true; + let connection_status_rx = ui.ipc().connection_status(); + let mut screen = MainScreen { elements, nav_grid, selected_coords: (1, 0), graphs_open, + connection_status_rx, }; screen.focus_current(); screen @@ -147,6 +153,40 @@ impl MainScreen { self.focus_current(); } + + /// Cycle focus between unique elements in the navigation grid. + fn navigate_focus(&mut self, forward: bool) { + // Collect unique elements in grid order. + let mut positions: Vec<(usize, usize)> = Vec::new(); // (row, col) + let mut seen: Vec> = Vec::new(); + for (y, row) in self.nav_grid.iter().enumerate() { + for (x, elem_opt) in row.iter().enumerate() { + if elem_opt.is_some() && !seen.contains(elem_opt) { + seen.push(*elem_opt); + positions.push((y, x)); + } + } + } + + let current = self.selected_coords; + let current_pos = positions + .iter() + .position(|&(r, c)| r == current.0 && c == current.1); + + let next_pos = if let Some(idx) = current_pos { + if forward { + (idx + 1) % positions.len() + } else { + (idx + positions.len() - 1) % positions.len() + } + } else { + 0 + }; + + self.unfocus_current(self.selected_coords.0, self.selected_coords.1); + self.selected_coords = positions[next_pos]; + self.focus_current(); + } } impl Screen for MainScreen { @@ -159,7 +199,21 @@ impl Screen for MainScreen { } fn render(&self, f: &mut Frame, rect: Rect) { - let main_block = Block::default().title("Main").borders(Borders::ALL); + let status = self.connection_status_rx.borrow(); + let status_text = match &*status { + IpcConnectionState::Connected => "Connected".to_string(), + IpcConnectionState::Connecting => "Connecting...".to_string(), + IpcConnectionState::Reconnecting { attempt } => { + format!("Reconnecting (attempt {})...", attempt) + } + IpcConnectionState::Incompatible { message } => { + format!("Incompatible: {}", message) + } + IpcConnectionState::Disconnected => "Disconnected".to_string(), + }; + let main_block = Block::default() + .title(format!("Main [{}]", status_text)) + .borders(Borders::ALL); f.render_widget(main_block, rect); let inner = rect.inner(Margin { @@ -215,10 +269,14 @@ impl Screen for MainScreen { fn handle_input(&mut self, event: KeyEvent) -> InteractionResult { match event.code { - KeyCode::Up => self.navigate(NavDirection::Up), - KeyCode::Down => self.navigate(NavDirection::Down), - KeyCode::Left => self.navigate(NavDirection::Left), - KeyCode::Right => self.navigate(NavDirection::Right), + KeyCode::Tab => { + self.navigate_focus(true); + return InteractionResult::Handled; + } + KeyCode::BackTab => { + self.navigate_focus(false); + return InteractionResult::Handled; + } KeyCode::Enter | KeyCode::Char(' ') if self.selected_coords.1 == 1 => { self.graphs_open = !self.graphs_open; for element in self.elements.iter_mut() { @@ -232,7 +290,17 @@ impl Screen for MainScreen { let (y, x) = self.selected_coords; if let Some(Some(index)) = self.nav_grid.get(y).and_then(|r| r.get(x)) { if let Some(el) = self.elements.get_mut(*index) { - return el.interact(event); + let result = el.interact(event); + if matches!(result, InteractionResult::Unhandled) { + match event.code { + KeyCode::Up => self.navigate(NavDirection::Up), + KeyCode::Down => self.navigate(NavDirection::Down), + KeyCode::Left => self.navigate(NavDirection::Left), + KeyCode::Right => self.navigate(NavDirection::Right), + _ => {} + } + } + return result; } } } diff --git a/iota-core/Cargo.toml b/iota-core/Cargo.toml index 3f7a5f6..06a58f0 100644 --- a/iota-core/Cargo.toml +++ b/iota-core/Cargo.toml @@ -24,3 +24,4 @@ pnet = "0.35.0" ratatui = "0.30.0" reqwest = "0.13.2" tokio = { version = "1.50.0", features = ["full"] } +tokio-util = { version = "0.7", features = ["rt"] } diff --git a/iota-core/src/main.rs b/iota-core/src/main.rs index b0a3bb6..35f9077 100644 --- a/iota-core/src/main.rs +++ b/iota-core/src/main.rs @@ -153,7 +153,7 @@ async fn main() { if !web_server::start(port).await { log!("Failed to start the MTP web server on port {}", port); } - let _ = omikron::omikron_connection::get_omikron_connection().await; + let _ = omikron::omikron_connection::get_omikron_connection(tokio_util::sync::CancellationToken::new()).await; log_t!("setup_completed"); loop { diff --git a/iota-daemon-lib/Cargo.toml b/iota-daemon-lib/Cargo.toml index abb2245..10fd67b 100644 --- a/iota-daemon-lib/Cargo.toml +++ b/iota-daemon-lib/Cargo.toml @@ -11,5 +11,9 @@ iota-storage = { path = "../iota-storage" } iota-util = { path = "../iota-util" } omikron-connector = { path = "../omikron-connector" } mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } +dashmap = "6.1.0" +libc = "0.2" sysinfo = "0.38.3" tokio = { version = "1.50.0", features = ["full"] } +tokio-util = { version = "0.7", features = ["rt"] } +uuid = { version = "*", features = ["v4"] } diff --git a/iota-daemon-lib/src/command_router.rs b/iota-daemon-lib/src/command_router.rs index 11e2ddd..53f4327 100644 --- a/iota-daemon-lib/src/command_router.rs +++ b/iota-daemon-lib/src/command_router.rs @@ -1,5 +1,7 @@ use crate::DaemonRuntime; -use iota_ipc::DaemonMessage; +use iota_ipc::{ + IpcErrorCode, LocalRequest, ResponseEnvelope, ResponseResult, +}; use iota_logger::{log, log_command}; use iota_storage::users::user_manager; use iota_storage::util::config_util::modify_config; @@ -8,6 +10,8 @@ use omikron_connector::omikron_connection::OMIKRON_CONNECTION; use std::sync::Arc; use std::time::Duration; +use crate::daemon_state::ShutdownReason; + #[derive(Clone)] pub struct CommandRouter { runtime: Arc, @@ -18,85 +22,124 @@ impl CommandRouter { Self { runtime } } - pub async fn route(&self, seq: u64, line: String) -> DaemonMessage { - log_command!("{}", line); - let result = self.execute(&line).await; - DaemonMessage::CommandResult { - seq, - success: result.is_ok(), - message: result.unwrap_or_else(|error| error), + pub async fn route(&self, request_id: u64, request: LocalRequest) -> ResponseEnvelope { + log_command!("{:?}", request); + let result = self.execute(request).await; + ResponseEnvelope { + request_id, + result, } } - async fn execute(&self, line: &str) -> Result { - let parts = line + /// Parse a legacy console command string into a typed request. + pub fn parse_console_command(line: &str) -> Option { + let parts: Vec<&str> = line .trim_start_matches('/') .split_whitespace() - .collect::>(); + .collect(); match parts.as_slice() { - ["tasks"] => Ok(self - .runtime - .state - .active_tasks - .iter() - .map(|task| task.to_string()) - .collect::>() - .join(", ")), - ["help"] => Ok( - "Available commands: tasks, ping, user, reconnect, regenerate, reload, shutdown" - .into(), - ), - ["ping"] => self.ping(20).await, - ["ping", seconds] => self.ping(seconds.parse::().unwrap_or(20)).await, - ["user", "add", username] => { - let (user, _) = omikron_connector::user_ops::create_user(username).await; - user.map(|user| format!("Created user {}", user.user_id)) - .ok_or_else(|| "User creation failed".into()) - } + ["help"] => None, + ["tasks"] => Some(LocalRequest::ListTasks), + ["ping", _] | ["ping"] => None, + ["user", "add", username] => Some(LocalRequest::CreateUser { + username: username.to_string(), + }), ["user", "remove", username] => { - let user = user_manager::get_user_by_username(username) - .ok_or_else(|| "Username does not exist".to_string())?; + let user = user_manager::get_user_by_username(username)?; + Some(LocalRequest::RemoveUser { + user_id: user.user_id, + }) + } + ["user", "list"] => Some(LocalRequest::ListUsers), + ["reconnect"] => Some(LocalRequest::ReconnectOmikron), + ["regenerate", "keys"] => Some(LocalRequest::RotateIotaIdentity), + ["reload"] | ["restart"] => Some(LocalRequest::RestartDaemon), + ["shutdown"] | ["stop"] => Some(LocalRequest::StopDaemon), + _ => None, + } + } + + async fn execute(&self, request: LocalRequest) -> ResponseResult { + match request { + LocalRequest::GetStatus => { + let phase = self.runtime.current_startup_phase(); + let degraded = self.runtime.degraded_reason.borrow().clone(); + let tasks: Vec = self + .runtime + .state + .active_tasks + .iter() + .map(|task| task.to_string()) + .collect(); + let mut info = format!("Phase: {:?}, Tasks: {}", phase, tasks.join(", ")); + if let Some(reason) = degraded { + info.push_str(&format!(", Degraded: {}", reason)); + } + ResponseResult::Ok(info) + } + LocalRequest::ListTasks => { + let tasks: Vec = self + .runtime + .state + .active_tasks + .iter() + .map(|task| task.to_string()) + .collect(); + ResponseResult::Ok(tasks.join(", ")) + } + LocalRequest::ListUsers => { + let users: Vec = user_manager::get_users() + .into_iter() + .map(|user| format!("{} ({})", user.username, user.user_id)) + .collect(); + ResponseResult::Ok(users.join("\n")) + } + LocalRequest::CreateUser { username } => { + match omikron_connector::user_ops::create_user(&username).await { + (Some(user), _) => { + ResponseResult::Ok(format!("Created user {}", user.user_id)) + } + _ => ResponseResult::Error(IpcErrorCode::StorageFailure), + } + } + LocalRequest::RemoveUser { user_id } => { + let user = match user_manager::get_user(user_id) { + Some(user) => user, + None => return ResponseResult::Error(IpcErrorCode::NotFound), + }; let message = CommunicationValue::new(CommunicationType::DeleteUser) .with_sender(user.user_id as u64); - OMIKRON_CONNECTION - .send_message(&message) - .await - .map_err(|error| error.to_string())?; + if let Err(_e) = OMIKRON_CONNECTION.send_message(&message).await { + return ResponseResult::Error(IpcErrorCode::OmikronUnavailable); + } user_manager::remove_user(user.user_id); - Ok(format!("Removed user {}", user.user_id)) + ResponseResult::Ok(format!("Removed user {}", user.user_id)) } - ["user", "list"] => Ok(user_manager::get_users() - .into_iter() - .map(|user| format!("{} ({})", user.username, user.user_id)) - .collect::>() - .join("\n")), - ["reconnect"] => { + LocalRequest::ReconnectOmikron => { OMIKRON_CONNECTION.reconnect().await; - Ok("Reconnected to Omikron server".into()) + ResponseResult::Ok("Reconnected to Omikron server".into()) } - ["regenerate", "keys"] => { + LocalRequest::RotateIotaIdentity => { modify_config(|config| { config.public_key = None; config.private_key = None; config.iota_id = None; }); OMIKRON_CONNECTION.reconnect().await; - Ok("Key pair regenerated and Omikron reconnection requested".into()) + ResponseResult::Ok("Key pair regenerated and Omikron reconnection requested".into()) } - ["reload"] | ["restart"] => { - *self.runtime.state.reload.write().await = true; - *self.runtime.state.shutdown.write().await = true; - Ok("Daemon restart requested".into()) + LocalRequest::RestartDaemon => { + self.runtime.shutdown(ShutdownReason::Restart); + ResponseResult::Ok("Daemon restart requested".into()) } - ["shutdown"] | ["stop"] => { - *self.runtime.state.shutdown.write().await = true; - Ok("Daemon shutdown requested".into()) + LocalRequest::StopDaemon => { + self.runtime.shutdown(ShutdownReason::Stop); + ResponseResult::Ok("Daemon shutdown requested".into()) } - _ => Err("Unknown command".into()), } } - async fn ping(&self, seconds: u64) -> Result { + pub async fn ping(&self, seconds: u64) -> Result { let response = OMIKRON_CONNECTION .await_response( &CommunicationValue::new(CommunicationType::Ping), diff --git a/iota-daemon-lib/src/daemon_state.rs b/iota-daemon-lib/src/daemon_state.rs index ef26e63..aa998b6 100644 --- a/iota-daemon-lib/src/daemon_state.rs +++ b/iota-daemon-lib/src/daemon_state.rs @@ -3,21 +3,123 @@ use iota_state::DaemonState; use std::sync::Arc; use std::time::Duration; use sysinfo::{RefreshKind, System}; +use tokio::sync::watch; +use tokio_util::sync::CancellationToken; + +/// Reason the daemon is shutting down. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ShutdownReason { + Stop, + Restart, + Fatal(String), +} + +impl ShutdownReason { + pub fn exit_code(&self) -> i32 { + match self { + ShutdownReason::Stop => 0, + ShutdownReason::Restart => 75, + ShutdownReason::Fatal(_) => 1, + } + } +} + +/// Tracks the lifecycle phase of the daemon for IPC visibility. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum StartupPhase { + Starting, + MigratingStorage, + LoadingUsers, + StartingServices, + Ready, + Degraded, + Stopping, +} + +impl From for iota_ipc::StartupPhase { + fn from(phase: StartupPhase) -> Self { + match phase { + StartupPhase::Starting => iota_ipc::StartupPhase::Starting, + StartupPhase::MigratingStorage => iota_ipc::StartupPhase::MigratingStorage, + StartupPhase::LoadingUsers => iota_ipc::StartupPhase::LoadingUsers, + StartupPhase::StartingServices => iota_ipc::StartupPhase::StartingServices, + StartupPhase::Ready => iota_ipc::StartupPhase::Ready, + StartupPhase::Degraded => iota_ipc::StartupPhase::Degraded, + StartupPhase::Stopping => iota_ipc::StartupPhase::Stopping, + } + } +} /* This wrapper exposes daemon state as IPC-safe snapshots while preserving a - * single owned state instance for all daemon subsystems. */ -#[derive(Clone, Default)] + * single owned state instance for all daemon subsystems. The cancellation token + * is the single lifecycle signal — all subsystems check it instead of a + * separate boolean. */ pub struct DaemonRuntime { pub state: Arc, + pub cancellation: CancellationToken, + pub shutdown_tx: watch::Sender>, + pub startup_phase: watch::Sender, + pub degraded_reason: watch::Sender>, +} + +impl Clone for DaemonRuntime { + fn clone(&self) -> Self { + Self { + state: self.state.clone(), + cancellation: self.cancellation.clone(), + shutdown_tx: self.shutdown_tx.clone(), + startup_phase: self.startup_phase.clone(), + degraded_reason: self.degraded_reason.clone(), + } + } +} + +impl Default for DaemonRuntime { + fn default() -> Self { + Self::new() + } } impl DaemonRuntime { pub fn new() -> Self { + let (shutdown_tx, _) = watch::channel(None); + let (startup_phase, _) = watch::channel(StartupPhase::Starting); + let (degraded_reason, _) = watch::channel(None); Self { state: Arc::new(DaemonState::new()), + cancellation: CancellationToken::new(), + shutdown_tx, + startup_phase, + degraded_reason, } } + pub fn shutdown(&self, reason: ShutdownReason) { + self.cancellation.cancel(); + let _ = self.shutdown_tx.send(Some(reason)); + } + + pub fn shutdown_reason(&self) -> Option { + self.shutdown_tx.borrow().clone() + } + + pub fn is_shutting_down(&self) -> bool { + self.cancellation.is_cancelled() + } + + pub fn set_startup_phase(&self, phase: StartupPhase) { + let _ = self.startup_phase.send(phase); + } + + pub fn current_startup_phase(&self) -> StartupPhase { + *self.startup_phase.borrow() + } + + pub fn mark_degraded(&self, reason: String) { + let _ = self.degraded_reason.send(Some(reason.clone())); + let _ = self.startup_phase.send(StartupPhase::Degraded); + } + pub fn snapshot(&self) -> StateSnapshot { let state = self .state @@ -41,7 +143,7 @@ impl DaemonRuntime { let mut system = System::new_with_specifics(RefreshKind::everything()); let mut counter = 0.0; loop { - if *runtime.state.shutdown.read().await { + if runtime.is_shutting_down() { break; } system.refresh_cpu_all(); diff --git a/iota-daemon-lib/src/ipc_server.rs b/iota-daemon-lib/src/ipc_server.rs index 11f62f4..2e3f6e9 100644 --- a/iota-daemon-lib/src/ipc_server.rs +++ b/iota-daemon-lib/src/ipc_server.rs @@ -1,28 +1,42 @@ use crate::{CommandRouter, DaemonRuntime}; -use iota_ipc::{ClientMessage, DaemonMessage, read_msg, write_msg}; +use iota_ipc::{ + ClientMessage, DaemonMessage, HelloAck, MIN_PROTOCOL_VERSION, PROTOCOL_VERSION, read_msg, + write_msg, +}; +use iota_logger::log; use std::io::Result; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::{env, os::fd::FromRawFd}; use tokio::net::{UnixListener, UnixStream}; -use tokio::sync::broadcast; +use tokio::sync::{broadcast, mpsc, watch}; +use uuid::Uuid; + +/// Per-client outbound queue capacity. +const CLIENT_CHANNEL_SIZE: usize = 256; + +/// Maximum handshake retries before giving up. +const MAX_HANDSHAKE_RETRIES: u32 = 10; pub struct IpcServer { path: PathBuf, runtime: Arc, - messages: broadcast::Sender, + log_tx: broadcast::Sender, + state_rx: watch::Sender, } impl IpcServer { pub fn new( path: impl Into, runtime: Arc, - messages: broadcast::Sender, + log_tx: broadcast::Sender, + state_rx: watch::Sender, ) -> Self { Self { path: path.into(), runtime, - messages, + log_tx, + state_rx, } } @@ -38,11 +52,14 @@ impl IpcServer { } }; loop { - let (stream, _) = listener.accept().await?; + let (stream, _addr) = listener.accept().await?; let runtime = self.runtime.clone(); - let messages = self.messages.clone(); + let log_tx = self.log_tx.clone(); + let state_rx = self.state_rx.clone(); tokio::spawn(async move { - let _ = handle_client(stream, runtime, messages).await; + if let Err(error) = handle_client(stream, runtime, log_tx, state_rx).await { + eprintln!("IPC client error: {error}"); + } }); } } @@ -72,34 +89,159 @@ async fn remove_stale_socket(path: &Path) -> Result<()> { } } +#[derive(Clone, Debug)] +struct PeerIdentity { + pid: i32, + uid: u32, + gid: u32, +} + +fn peer_credentials(stream: &UnixStream) -> PeerIdentity { + #[cfg(target_os = "linux")] + { + use std::os::unix::io::AsRawFd; + unsafe { + let mut cred: libc::ucred = std::mem::zeroed(); + let mut len = std::mem::size_of::() as libc::socklen_t; + let fd = stream.as_raw_fd(); + libc::getsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_PEERCRED, + &mut cred as *mut _ as *mut libc::c_void, + &mut len, + ); + PeerIdentity { + pid: cred.pid, + uid: cred.uid, + gid: cred.gid, + } + } + } + #[cfg(not(target_os = "linux"))] + { + PeerIdentity { + pid: 0, + uid: 0, + gid: 0, + } + } +} + async fn handle_client( stream: UnixStream, runtime: Arc, - messages: broadcast::Sender, + log_tx: broadcast::Sender, + _state_rx: watch::Sender, ) -> Result<()> { + let peer = peer_credentials(&stream); let (mut reader, mut writer) = stream.into_split(); - let mut outgoing = messages.subscribe(); - let initial = DaemonMessage::StateUpdate(runtime.snapshot()); - write_msg(&mut writer, &initial).await?; - let writer_task = tokio::spawn(async move { - while let Ok(message) = outgoing.recv().await { - if write_msg(&mut writer, &message).await.is_err() { + let (directed_tx, directed_rx) = mpsc::channel::(CLIENT_CHANNEL_SIZE); + + // --- Handshake --- + let mut negotiated_version: Option = None; + for _ in 0..MAX_HANDSHAKE_RETRIES { + match read_msg::<_, ClientMessage>(&mut reader).await { + Ok(ClientMessage::Hello { supported_versions }) => { + let version = supported_versions + .iter() + .copied() + .find(|v| *v >= MIN_PROTOCOL_VERSION && *v <= PROTOCOL_VERSION) + .unwrap_or(PROTOCOL_VERSION); + negotiated_version = Some(version); + let instance_id = Uuid::new_v4().to_string(); + let ack = DaemonMessage::HelloAck(HelloAck { + protocol_version: version, + daemon_version: env!("CARGO_PKG_VERSION").to_string(), + instance_id, + startup_phase: runtime.current_startup_phase().into(), + capabilities: vec!["commands".into(), "metrics".into(), "logs".into()], + }); + write_msg(&mut writer, &ack).await?; break; } + Ok(_) => { + // Unexpected first message — send error and close. + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Expected Hello as first message", + )); + } + Err(e) => return Err(e), } - }); + } + let _version = negotiated_version.ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::Other, "Handshake failed after retries") + })?; + + log!("IPC client connected (pid={}, uid={})", peer.pid, peer.uid); + + // --- Send initial state snapshot --- + let initial = DaemonMessage::StateUpdate(runtime.snapshot()); + let _ = directed_tx.send(initial).await; + + // --- Writer task: merge directed responses + shared log events --- + let mut log_rx = log_tx.subscribe(); + let directed_for_writer = directed_tx.clone(); + let writer_task = { + let runtime = runtime.clone(); + tokio::spawn(async move { + let mut directed_rx = directed_rx; + loop { + tokio::select! { + // Directed messages (responses to this client's requests) + msg = directed_rx.recv() => { + match msg { + Some(message) => { + if write_msg(&mut writer, &message).await.is_err() { + break; + } + } + None => break, + } + } + // Shared log events + result = log_rx.recv() => { + match result { + Ok(message) => { + if write_msg(&mut writer, &message).await.is_err() { + break; + } + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + let _ = directed_for_writer.send(DaemonMessage::Gap { skipped }).await; + // Then send current snapshot for resync + let _ = directed_for_writer.send( + DaemonMessage::StateUpdate(runtime.snapshot()) + ).await; + } + Err(broadcast::error::RecvError::Closed) => break, + } + } + } + } + }) + }; + + // --- Reader loop --- let router = CommandRouter::new(runtime.clone()); loop { match read_msg::<_, ClientMessage>(&mut reader).await { - Ok(ClientMessage::Command { seq, line }) => { - let result = router.route(seq, line).await; - let _ = messages.send(result); + Ok(ClientMessage::Request(envelope)) => { + let response = router.route(envelope.request_id, envelope.request).await; + let _ = directed_tx.send(DaemonMessage::Response(response)).await; } - Ok(ClientMessage::Subscribe) => { - let _ = messages.send(DaemonMessage::StateUpdate(runtime.snapshot())); + Ok(ClientMessage::Subscribe { .. }) => { + let snapshot = DaemonMessage::StateUpdate(runtime.snapshot()); + let _ = directed_tx.send(snapshot).await; } Ok(ClientMessage::Ping { seq }) => { - let _ = messages.send(DaemonMessage::Pong { seq }); + let _ = directed_tx.send(DaemonMessage::Pong { seq }).await; + } + Ok(ClientMessage::Hello { .. }) => { + // Re-handshake on existing connection: treat as resubscribe + let snapshot = DaemonMessage::StateUpdate(runtime.snapshot()); + let _ = directed_tx.send(snapshot).await; } Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => break, Err(error) => { @@ -109,5 +251,10 @@ async fn handle_client( } } writer_task.abort(); + log!( + "IPC client disconnected (pid={}, uid={})", + peer.pid, + peer.uid + ); Ok(()) } diff --git a/iota-daemon-lib/src/lib.rs b/iota-daemon-lib/src/lib.rs index 7f10773..48f8c68 100644 --- a/iota-daemon-lib/src/lib.rs +++ b/iota-daemon-lib/src/lib.rs @@ -4,5 +4,5 @@ pub mod ipc_server; pub mod log_broadcaster; pub use command_router::CommandRouter; -pub use daemon_state::DaemonRuntime; +pub use daemon_state::{DaemonRuntime, ShutdownReason, StartupPhase}; pub use ipc_server::IpcServer; diff --git a/iota-daemon/Cargo.toml b/iota-daemon/Cargo.toml index e5d5864..620b413 100644 --- a/iota-daemon/Cargo.toml +++ b/iota-daemon/Cargo.toml @@ -12,3 +12,4 @@ iota-storage = { path = "../iota-storage" } omikron-connector = { path = "../omikron-connector" } web-server = { path = "../web-server" } tokio = { version = "1.50.0", features = ["full"] } +tokio-util = { version = "0.7", features = ["rt"] } diff --git a/iota-daemon/src/main.rs b/iota-daemon/src/main.rs index 2cdb8ff..b9404f1 100644 --- a/iota-daemon/src/main.rs +++ b/iota-daemon/src/main.rs @@ -1,12 +1,11 @@ -use iota_daemon_lib::{DaemonRuntime, IpcServer, log_broadcaster}; +use iota_daemon_lib::{DaemonRuntime, IpcServer, ShutdownReason, StartupPhase, log_broadcaster}; use iota_logger::{self as logger, log, log_t}; use iota_storage::users::user_manager; use iota_storage::util::config_util::CONFIG; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; -use tokio::sync::broadcast; - +use tokio::sync::{broadcast, watch}; fn socket_path() -> PathBuf { std::env::var_os("IOTA_SOCKET") .map(PathBuf::from) @@ -17,42 +16,79 @@ fn socket_path() -> PathBuf { async fn main() { logger::startup(); iota_storage::util::config_util::load_config(); + + let runtime = Arc::new(DaemonRuntime::new()); + runtime.set_startup_phase(StartupPhase::LoadingUsers); + if user_manager::load_users().await.is_err() { log_t!("user_load_failed"); } - let runtime = Arc::new(DaemonRuntime::new()); + + // --- IPC infrastructure --- + let (log_tx, _) = broadcast::channel(512); + log_broadcaster::spawn(log_tx.clone()); + let (state_tx, _state_rx) = watch::channel(iota_ipc::StateSnapshot::default()); + + // --- Start IPC server early (before services) so clients can see startup phases --- + runtime.set_startup_phase(StartupPhase::StartingServices); + let ipc_server = IpcServer::new( + socket_path(), + runtime.clone(), + log_tx.clone(), + state_tx.clone(), + ); + tokio::spawn(async move { + if let Err(error) = ipc_server.run().await { + eprintln!("iota-daemon IPC server failed: {error}"); + } + }); + log!("iota-daemon IPC server started"); + + // --- System monitor --- runtime.spawn_system_monitor(); - let (messages, _) = broadcast::channel(512); - log_broadcaster::spawn(messages.clone()); - let state_updates = runtime.clone(); - let state_messages = messages.clone(); + + // --- State update publisher (watch-based, no full broadcast per tick) --- + let state_publisher = runtime.clone(); tokio::spawn(async move { loop { - if *state_updates.state.shutdown.read().await { + if state_publisher.is_shutting_down() { break; } - let _ = state_messages.send(iota_ipc::DaemonMessage::StateUpdate( - state_updates.snapshot(), - )); + let snapshot = state_publisher.snapshot(); + let _ = state_tx.send(snapshot); tokio::time::sleep(Duration::from_millis(500)).await; } }); + // --- Web server --- let port = CONFIG.load().port; - if !web_server::start(port).await { + if !web_server::start(port, runtime.cancellation.clone()).await { log!("Failed to start the MTP web server on port {}", port); + runtime.mark_degraded("MTP web server failed to start".into()); } - let _ = omikron_connector::omikron_connection::get_omikron_connection().await; - let server = IpcServer::new(socket_path(), runtime.clone(), messages); - tokio::spawn(async move { - if let Err(error) = server.run().await { - eprintln!("iota-daemon IPC server failed: {error}"); - } - }); - log!("iota-daemon started"); - while !*runtime.state.shutdown.read().await { - tokio::time::sleep(Duration::from_millis(250)).await; + // --- Omikron connection --- + let omikron_result = + omikron_connector::omikron_connection::get_omikron_connection(runtime.cancellation.clone()) + .await; + if omikron_result.is_none() { + runtime.mark_degraded("Omikron connection unavailable".into()); } - log!("iota-daemon stopping"); + + runtime.set_startup_phase(StartupPhase::Ready); + log!("iota-daemon started (phase: Ready)"); + + // --- Main lifecycle loop --- + runtime.cancellation.cancelled().await; + + let reason = runtime.shutdown_reason().unwrap_or(ShutdownReason::Stop); + log!("iota-daemon shutting down (reason: {:?})", reason); + runtime.set_startup_phase(StartupPhase::Stopping); + + // Wait a moment for in-flight operations to complete + tokio::time::sleep(Duration::from_millis(500)).await; + + let exit_code = reason.exit_code(); + log!("iota-daemon exited (code: {})", exit_code); + std::process::exit(exit_code); } diff --git a/iota-ipc/src/lib.rs b/iota-ipc/src/lib.rs index 8cbd6fa..11105c2 100644 --- a/iota-ipc/src/lib.rs +++ b/iota-ipc/src/lib.rs @@ -1,5 +1,14 @@ pub mod protocol; pub mod transport; -pub use protocol::{ClientMessage, DaemonMessage, LogEntry, StateSnapshot}; +pub use protocol::{ + ClientMessage, DaemonMessage, HelloAck, LogEntry, StateSnapshot, MetricSample, + RequestEnvelope, ResponseEnvelope, ResponseResult, LocalRequest, IpcErrorCode, + ConnectionStatus, StartupPhase, LifecycleEvent, +}; pub use transport::{read_msg, write_msg}; + +/// Current IPC protocol version. +pub const PROTOCOL_VERSION: u16 = 2; +/// Minimum protocol version this daemon understands. +pub const MIN_PROTOCOL_VERSION: u16 = 2; diff --git a/iota-ipc/src/protocol.rs b/iota-ipc/src/protocol.rs index 09a723b..35a93d6 100644 --- a/iota-ipc/src/protocol.rs +++ b/iota-ipc/src/protocol.rs @@ -1,28 +1,123 @@ use serde::{Deserialize, Serialize}; +// --------------------------------------------------------------------------- +// Client → Daemon +// --------------------------------------------------------------------------- + #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(tag = "type", content = "data", rename_all = "snake_case")] pub enum ClientMessage { - Command { seq: u64, line: String }, - Subscribe, + Hello { supported_versions: Vec }, + Subscribe { log_classes: Vec, metric_interval_ms: Option }, + Request(RequestEnvelope), Ping { seq: u64 }, } +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct RequestEnvelope { + pub request_id: u64, + pub protocol_version: u16, + pub request: LocalRequest, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] +pub enum LocalRequest { + GetStatus, + ListTasks, + ListUsers, + CreateUser { username: String }, + RemoveUser { user_id: i64 }, + ReconnectOmikron, + RotateIotaIdentity, + RestartDaemon, + StopDaemon, +} + +// --------------------------------------------------------------------------- +// Daemon → Client +// --------------------------------------------------------------------------- + #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(tag = "type", content = "data", rename_all = "snake_case")] pub enum DaemonMessage { + HelloAck(HelloAck), LogEntry(LogEntry), StateUpdate(StateSnapshot), - CommandResult { - seq: u64, - success: bool, - message: String, - }, - Pong { - seq: u64, - }, + MetricSample(MetricSample), + Response(ResponseEnvelope), + Pong { seq: u64 }, + LifecycleEvent(LifecycleEvent), + Gap { skipped: u64 }, } +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct HelloAck { + pub protocol_version: u16, + pub daemon_version: String, + pub instance_id: String, + pub startup_phase: StartupPhase, + pub capabilities: Vec, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct ResponseEnvelope { + pub request_id: u64, + pub result: ResponseResult, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] +pub enum ResponseResult { + Ok(String), + Error(IpcErrorCode), +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum IpcErrorCode { + InvalidRequest, + NotFound, + Conflict, + StorageFailure, + OmikronUnavailable, + UnsupportedVersion, + NotReady, + Disconnected, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] +pub enum LifecycleEvent { + StateChanged(ConnectionStatus), + Shutdown { reason: String }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ConnectionStatus { + Connected, + Reconnecting, + Degraded, + Disconnected, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StartupPhase { + Starting, + MigratingStorage, + LoadingUsers, + StartingServices, + Ready, + Degraded, + Stopping, +} + +// --------------------------------------------------------------------------- +// Shared types +// --------------------------------------------------------------------------- + #[derive(Clone, Debug, Deserialize, Serialize)] pub struct LogEntry { pub timestamp_ms: u128, @@ -40,3 +135,12 @@ pub struct StateSnapshot { pub net_down: Vec<(f64, f64)>, pub sys_info: String, } + +#[derive(Clone, Debug, Default, Deserialize, Serialize)] +pub struct MetricSample { + pub cpu: Option, + pub ram: Option, + pub ping: Option, + pub net_up: Option, + pub net_down: Option, +} diff --git a/iota-ipc/src/transport.rs b/iota-ipc/src/transport.rs index 811b0ee..2793451 100644 --- a/iota-ipc/src/transport.rs +++ b/iota-ipc/src/transport.rs @@ -41,19 +41,20 @@ where #[cfg(test)] mod tests { use super::{read_msg, write_msg}; - use crate::protocol::ClientMessage; + use crate::protocol::{ClientMessage, LocalRequest, RequestEnvelope}; #[tokio::test] async fn round_trips_framed_messages() { let (mut writer, mut reader) = tokio::io::duplex(1024); - let message = ClientMessage::Command { - seq: 4, - line: "help".into(), - }; + let message = ClientMessage::Request(RequestEnvelope { + request_id: 4, + protocol_version: 2, + request: LocalRequest::GetStatus, + }); write_msg(&mut writer, &message) .await .expect("write succeeds"); let received: ClientMessage = read_msg(&mut reader).await.expect("read succeeds"); - assert!(matches!(received, ClientMessage::Command { seq: 4, line } if line == "help")); + assert!(matches!(received, ClientMessage::Request(req) if req.request_id == 4)); } } diff --git a/iota.mk b/iota.mk deleted file mode 100644 index 61083565cb37fe02c651ab466a776954201300aa..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 5713 zcmeYb@%3h8Il$J__iMKL@(V}b2FX1>u>2;wk#B{RyrH;gfyv2GwW24j2Rh~Ff62PN zl-;AW*+__G;?!Ng7Oq&OAgP$C6gi=9Mn)^^iOwIH(&t<9{|nlGHD@|~Vvqf`9V<_) zU^v6b%3bM}vGYst2BGiO=Xcr^8s()dSW?|?xZ~26$3pIU*ZCG-i}KkuvsG)2<|#Wd z?o5qvwTIK{*)E(o_r32^jIZ$IzGIt8S6@`*HQx673un0L?9*SEbtKvDHg+vm+$@-y zov7Kz|H5PD!lc%nP1)VzvqDo=|4L}!VsXwm@^8`k4ZmzGUo|@|2|Kv!46A!Z>xbT# zU#3TQT9%w#>wakGgOr&eJNJ7Xm5}GQi*%Ftykc3k7LVfTEj!F*TSV$aTHNOU?7gri z@zJ)r8=O-QKJ}H5w4bxGDemL0-)~lQN6&iSlRGi1Z3S}!>nDrO8#}Ho4%JRE?fxaX zNKt!JwC>{+zvn%DD^`}=G~AbAsCn?&J}z#Z{d4%31x`#Z*d*h|Glj9Hbc5mao9q@R zYs6PI8_uzL$|0ET&iP#+>~x~A+frwX(+7+nu=s7Y5msCG^`WTk%=7s-;~q!vZaUh~ z6PdWpyu_numXoU3yTiKa5zZe?5_LXBl^kNY<>?!D+)j)wJcwh%|Cxu>rYI+I$Ndlu zNT}Iy=1o=2f|ULy7r55?X zLw`7ATttr@cGq6!xlwYtomq?8-@*f%Rj2kO> zuVoP@oJ>ACD;@eCeLGu=egGX zwwIx9im&oED`sAcX)^n|%ug!ZiHjBLwy`d*nK>R#uuyLIt%qvTh+Y@LPb zXO&Do&>U1|C*;<5Qlmqxu~KfKUZ})5;nm&C=5$Q33BGQ*>pGhV`-knDH?3LqxoVz6 z%-qxKYtNkdAHgtxs`so96{$^rOATX1Yh*vMH3b|Gop?-e!t6e#3JKlor}szY&-4kZ zQsLK3Pi53B7)-lr#?3}?Q8uI3)#pe)}9Y338xmz+0 zXVu)9-?XunA*o?`__|{z3v~|*{*Gaa6jxeOT=sv0^Zgn#&M#ior#2crVbsb@oBT#z z{-ysfb3XZMP5G{})jNWZEep>R?Dv0l=)!~O=f4@=GEd|=xKrF=52u&o`a8+(tZ64A z``c$`UWs1mW;4Ck`l!(7BdrR#av#LQF7$h3Dd_D{O<2y)F=LbSX3o1FsTCXMGF3-yB+tn8OqEiUk6{F7L-&jVAt#QgX}aO6 zh2MM*`L)ka;!a80F?}-Qgx?A=`x}?&T5Q<3pz!mHy($)_he{@t9P|3a;c{sb`|p}X zOOGn-`BnQ>X^GS7=$xk2I$08dtb#>9msprvgr1sxcfpF*RlV0|Zt-u6y>KM)L**J< zo8wx~r>(oGy><(KN~*iyVu!spT$$FD+!IX}o38ARkof*$=eos7y;C`!ePP+gp`&zM zc#>j5@=s-_&dhS3_%FM}ewlc5O|A%%F=L#$<=>2Tr*fLLBF^8a%CYrUo;#)YREOil zUAB$URz^Qqf}~}d(i=ZY?P*cUH$$A@yk|C(oU9UT21k@M?RHt1DU_WKF+RW%@@?Dp@*e$B}8061AG~UrTl# zDPnOsYJLAjlCePj{s|sC+~f|ZY+SczS@1U>mf4%5zv@^l^lYCxqo{p~eM7=R$GLZS z486E~WMb1xIacC|efI?}EYvzz!_j#RH;OSn_fcqU$`Y;K*n z-C~ve`!zUj8`W(Tt ztPizx@%-nS{JMI@l+M8Gzxg8-4upR;nX_XTUk=xuFOuJryvr01tWaHi?R#t0(i2QU zF~_=RhAnPqIKC^!E9lEola$F5qzikpe(zH;S-C?bhVj?-^K*OyZ}JpHnmxKUo3)|H ztw`e2QUr#uVewtBp)RLpyB;uyJRR#JS1m)jZ1Dd!CmBu?7C53&4av$Q3Aca4edZW~G6fOGpl zsXyAV$X@$c@Cw7-=BpMhwtbYmnkhZMb@hrXY$gV)Wq+(}zc{1x%*MK{-w) z3R@pet?|CHfiro6$y4hi?Ay;ex=i|Nd8ycc?e6oEEr(_pa=bLMd|uS}uVepJMX9IU zT^}o@6&DL$*kJJgcH!n|`{iGE_jkmoIEVbJP`7t>y{N?X$|iEDlFp?X=7*W=yrn1S zyE$o>b+ek*Bz&K)q`~m@QNpiN6&KUlxIOe|XfaH8%6o7vdEM28Q&?uoOpkneZ0A*n z=(*EoZT~5Js8L_x#e}9Kfv+cVF;*>|>Ah(FVu`-enY(t`b1R!Ie7Q0GhM*UBPq2WA zpfrz@jG*fiqgCbWU5+g|TClbGdHvlBh2As7-(QV8;#Ko#&Z2EI8CR}4Z4%BNx>>hs z%kHm2D@I#Tqb;b>7Sw19YOuDT-mmV~7YSS^6c^z-&FA`sJ+9(~i+Rp-G;b@bo*rOl z5iA~P_K&5S`}xv)Wr5qPZa)@t{qmSqJD5-Kj^lz^+BjH14rlMN2`sA-d zmv&9ODJ^zW-j?Y(myGUq&8kSL>hDravwrXOs?oapd|AA0%Gr-$cI}?~{{(TVa@S0+ zXWSosTZh|nt#kLG7bS^j&h9lxNKO}))7>WV%j&jr?&*I@3UijmalX9VUc(W3ibeCM z;J))ZOtvn4>fC>uIWKgcTBN*MM31TGxXjekJ%?*0K15C_tcmPfnY(h2$uyaD5=Fs< z?s1#CbsfZ3?%VJr*2L?TXp+tJh1~+00q&E1woFm6-fPagUoUyGkemAD*P?uxM*V8~ z89&?q&HlW@Na>8{jfo=lHmfveR^-lmvF4_u;#a@wRnGCM=Wg|X?!MhuaCvRplB(F- znr}9i-(G$0Szn(@ZH*X%s>5#a->#X_JSq1DVmBl{vlB8~vD3HwU4qrCr~P7r=FRH0 z%YJ=hJDHr6u6|GMhSvq2(Byl!as#)mjLBWT`Nvy@kKf&HzVvK3U7mG#p|u64O~=;n zEPk#tUfBFvHFd@Y@5ptx@>t^UCaSEB479)d)m>$mSZkL$$>586wXn zDqW4ptg+t|@j(2j;{)!0sXP5e3+sag-#Xe$_6AnxR-E&?{&U&RtyfoFWSW>AllDw~ zrEJrJ9QG-9Y!_Rz7S{=jZLU6C9c3YBqTQgo@AA6frN=Dad2HtK`ns_;jqg`fid3+H z^rz4}noh@qJe{^r+uc6to7fYUtB;N++Ir8r!SduEgQ$$|ve#Mug!8`z9v+O$j1WZbjN9ylc7I&N?%GSamhzYZ2r7c+Jh5 zO!xy9v~JSz)!b=aZ&0+>wpbi-TcY&6&nw3D(ksK#E`D$RU$JW8 zx8@~zjl!1m|LL-a5Bptnm|OhxgH25TV|i5U7`NTa-*~NHZiy_P&h0-h z*M&K0KHh$wSGKqN(9;On<2KucoD|L2uQ+O|H!sl3Uhg|;uiotHTa4e>l$sJX?{X?0 z{T#d5bc;COYsJjXuF}ig)T_=d3e>JokPrVAQ|O`Q@^t6qSBBS)OD}WvI1{&Tm+wJu zW$Q0BqOB78e9O1liMIMx_*Yw{$v;WhwCLB0sRw@QzZF+>7btByd8HvCk$L93-~je* z4UB1BuXk^qzxKR;W_Qu6jZ3deu2)ydcE8StW=MUSnEU`(t zzQK$K7Asr6pT6(8j^I{<={^c|D(969-n|Gq|8swa%b$-j9(&LJ{qlSA)SABC+DtVc zl9~f}IwfDSCO=e)sMQuseqQ{?>H5va57~nLw`Je%&hv4N@?i@W3etSlUbMSLesl54 zCDMm?-<2$=@_Cn&wdT*t+6+~nqZcC9t#+5YRdkPgcBW)<&O!cbGEQkHCLar6Y)V#b z`Fv{84W?!4Uk^6UeAlS3Y)i#q=N0SZ?*7ZZ78c-Hl5Qw}x%h!a!MeXcUcXmRoxFD6 z<_puF?3)oSJJTuZ_MWASCLgYTF=gk}R|kKZaoDZdv9rhSs%vqG?xw!%Rb>SsS5EG| zofz`PHwMAdHq;JU1^*A0Y zaq8PPj?EMHtt zS3bA=X&>FjQ8=mH@712aa@W>OvU&DvO(t)vXyviV7k+fwCQQgZp8NQf;|7;rW|rtV z3yO+et7RLf#m?W%`Txh~ZHJ2+qY7V75&poyQsMezMd8H{tj?>}*&1z_&?U&A5c*L5 ze)Se*yP4eUA382w?G@we`}Am8$Unt!) PathBuf { #[tokio::main(flavor = "multi_thread")] async fn main() { let path = socket_path(); - let ipc = match IpcClient::connect(&path).await { + let ipc = match IpcClient::connect_or_activate(&path).await { Ok(client) => client, Err(error) => { eprintln!( "Cannot connect to iota-daemon at {}: {error}", path.display() ); + eprintln!("Ensure iota-daemon.socket is enabled or iota-daemon is running."); std::process::exit(1); } }; + ipc.spawn_reconnector(); let ui = start_tui(ipc); ui.set_screen(Box::new(MainScreen::new(ui.clone()).await)) .await; diff --git a/omikron-connector/Cargo.toml b/omikron-connector/Cargo.toml index f083794..7526c63 100644 --- a/omikron-connector/Cargo.toml +++ b/omikron-connector/Cargo.toml @@ -19,6 +19,7 @@ dashmap = "6.2.1" json = "*" reqwest = "0.13.2" tokio = { version = "1.50.0", features = ["full"] } +tokio-util = { version = "0.7", features = ["rt"] } uuid = { version = "*", features = ["v4"] } base64 = "0.22.1" hex = "*" diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index e68a12f..45257c2 100755 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -1,6 +1,6 @@ use dashmap::DashMap; use iota_logger::{log, log_cv_in, log_cv_out, log_t}; -use iota_state::{ACTIVE_TASKS, SHUTDOWN}; +use iota_state::ACTIVE_TASKS; use iota_storage::util::chat_files::{self, MessageState, change_message_state}; use iota_storage::util::config_util::{CONFIG, modify_config}; use iota_storage::util::e2ee_storage::{self, PendingChatSecretForward, StoredChatSecret}; @@ -16,6 +16,7 @@ use std::time::{Duration, Instant}; use tokio::sync::{Mutex, RwLock, Semaphore, oneshot, watch}; use tokio::task::JoinHandle; use tokio::time::sleep; +use tokio_util::sync::CancellationToken; use uuid::Uuid; use crate::omega_discovery; @@ -161,10 +162,15 @@ pub struct OmikronConnection { pub app_sessions: Arc>, pub(crate) missed_pongs: Arc, handler_semaphore: Arc, + cancellation: CancellationToken, } impl OmikronConnection { pub fn new() -> Self { + Self::with_cancellation(CancellationToken::new()) + } + + pub fn with_cancellation(cancellation: CancellationToken) -> Self { let (shutdown_tx, _) = watch::channel(false); let (state_watch_tx, _) = watch::channel(ConnectionState::Disconnected); @@ -183,6 +189,7 @@ impl OmikronConnection { app_sessions: Arc::new(DashMap::new()), missed_pongs: Arc::new(AtomicU32::new(0)), handler_semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_HANDLERS)), + cancellation, } } @@ -237,7 +244,7 @@ impl OmikronConnection { } if let Some(sender) = self.sender.read().await.as_ref() { - sender.close(); + sender.close().await; } self.set_state(ConnectionState::Disconnected).await; @@ -250,7 +257,7 @@ impl OmikronConnection { let mut shutdown_rx = shutdown_rx; loop { - if *shutdown_rx.borrow() || *SHUTDOWN.read().await { + if *shutdown_rx.borrow() || self.cancellation.is_cancelled() { log_t!("omikron_connection_loop_shutdown"); break; } @@ -621,7 +628,7 @@ impl OmikronConnection { self.missed_pongs.load(Ordering::Relaxed) ); if let Some(sender) = self.sender.read().await.as_ref() { - sender.close(); + sender.close().await; } break; } @@ -1750,7 +1757,7 @@ impl OmikronConnection { if !sender.is_open() { drop(sender_guard); if let Some(sender) = self.sender.write().await.take() { - sender.close(); + sender.close().await; } self.fail_all_waiting_tasks(format!( "Send failed: connection closed (connection_id={})", @@ -1933,11 +1940,12 @@ pub static OMIKRON_CONNECTION: LazyLock> = LazyLock::new( conn }); -pub async fn get_omikron_connection() -> Arc { - let conn = OMIKRON_CONNECTION.clone(); - +pub async fn get_omikron_connection( + cancellation: CancellationToken, +) -> Option> { + let conn = Arc::new(OmikronConnection::with_cancellation(cancellation)); conn.connect().await; - conn + Some(conn) } impl iota_connection::connection_handler::ConnectionHandler for OmikronConnection { diff --git a/omikron-connector/src/user_ops.rs b/omikron-connector/src/user_ops.rs index d02ffbe..6d52dd5 100644 --- a/omikron-connector/src/user_ops.rs +++ b/omikron-connector/src/user_ops.rs @@ -1,6 +1,5 @@ use base64::{Engine as _, engine::general_purpose::STANDARD}; use iota_logger::{PrintType, log, log_cv, log_t}; -use iota_state::{RELOAD, SHUTDOWN}; use iota_storage::users::user_manager::{add_user, save_users}; use iota_storage::users::user_profile::UserProfile; use iota_util::crypto_helper::{self, hex_hash, public_key_bundle_to_base64}; @@ -81,8 +80,6 @@ pub async fn create_user(username: &str) -> (Option, Option log_t!("User creation: Response returned none"); return (None, None); } - *SHUTDOWN.write().await = true; - *RELOAD.write().await = true; log!("Created User"); save_file( "", diff --git a/systemd/iota-daemon.service b/systemd/iota-daemon.service index 3f3b819..b61e5e2 100644 --- a/systemd/iota-daemon.service +++ b/systemd/iota-daemon.service @@ -5,12 +5,22 @@ Wants=network-online.target Requires=iota-daemon.socket [Service] -Type=simple +Type=notify ExecStart=/usr/bin/iota-daemon Restart=on-failure +RestartSec=5s RuntimeDirectory=iota RuntimeDirectoryMode=0750 Environment=IOTA_SOCKET=/run/iota/iota.sock +# Exit code 75 = restart requested (daemon-specific convention) +RestartPreventExitStatus=0 +RestartForceExitStatus=75 + +# Graceful shutdown +TimeoutStopSec=10 +KillMode=mixed +KillSignal=SIGTERM + [Install] WantedBy=multi-user.target diff --git a/systemd/iota-daemon.socket b/systemd/iota-daemon.socket index 22ab691..b06a209 100644 --- a/systemd/iota-daemon.socket +++ b/systemd/iota-daemon.socket @@ -6,6 +6,7 @@ ListenStream=/run/iota/iota.sock SocketMode=0660 SocketUser=iota SocketGroup=iota +Backlog=5 RemoveOnStop=true [Install] diff --git a/web-server/Cargo.toml b/web-server/Cargo.toml index 9827408..be6b65a 100644 --- a/web-server/Cargo.toml +++ b/web-server/Cargo.toml @@ -8,7 +8,7 @@ edition = "2024" mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = ["web-server"] } bytes = "1" http = "1" -iota-state = { path = "../iota-state" } -iota-util = { path = "../iota-util" } iota-logger = { path = "../iota-logger" } +iota-util = { path = "../iota-util" } tokio = { version = "1.50.0", features = ["full"] } +tokio-util = { version = "0.7", features = ["rt"] } diff --git a/web-server/src/lib.rs b/web-server/src/lib.rs index 4dc18ab..f5214ab 100644 --- a/web-server/src/lib.rs +++ b/web-server/src/lib.rs @@ -1,11 +1,10 @@ use bytes::Bytes; use iota_logger::log; -use iota_state::{ACTIVE_TASKS, SHUTDOWN}; use iota_util::file_util::load_file_vec; use mtp::host::HostConfig; use mtp::webserver::{Http3Request, Http3Response, MTPWebServer, WebServerConfig}; use std::net::{IpAddr, Ipv4Addr}; -use tokio::time::{Duration, sleep}; +use tokio_util::sync::CancellationToken; const CERT_PATH: &str = "certs/cert.pem"; const KEY_PATH: &str = "certs/cert.key"; @@ -63,7 +62,7 @@ fn content_type(name: &str) -> &'static str { } } -pub async fn start(port: u16) -> bool { +pub async fn start(port: u16, cancellation: CancellationToken) -> bool { let certificate = match tokio::fs::read(CERT_PATH).await { Ok(certificate) => certificate, Err(error) => { @@ -102,7 +101,6 @@ pub async fn start(port: u16) -> bool { log!("MTP web server running on port {}", port); tokio::spawn(async move { - ACTIVE_TASKS.insert("WebServer".into()); loop { tokio::select! { result = server.accept() => { @@ -112,22 +110,12 @@ pub async fn start(port: u16) -> bool { Err(error) => log!("MTP webserver connection failed: {}", error), } } - _ = wait_for_shutdown() => { + _ = cancellation.cancelled() => { server.shutdown().await; break; } } } - ACTIVE_TASKS.remove("WebServer"); }); true } - -async fn wait_for_shutdown() { - loop { - if *SHUTDOWN.read().await { - break; - } - sleep(Duration::from_millis(100)).await; - } -}