1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
|
"""Entry point for the meshbay-hub systemd service, and its few CLI chores."""
import argparse
import asyncio
import logging
import sys
from pathlib import Path
import uvicorn
from meshbay_hub.config import load_config
def main() -> None:
parser = argparse.ArgumentParser(description="MeshBay Hub server")
parser.add_argument("--config", type=Path, default=None)
parser.add_argument("--log-level", default="INFO",
choices=["DEBUG", "INFO", "WARNING", "ERROR"])
# Optional on purpose: the systemd unit runs `meshbay-hub --config …` with no
# subcommand and must go on starting the server.
sub = parser.add_subparsers(dest="command")
prune = sub.add_parser(
"prune-groups",
help="delete groups no node ever hosted (meant for cron)")
prune.add_argument("--days", type=int, default=None,
help="grace period since creation (default 7)")
prune.add_argument("--dry-run", action="store_true",
help="list what would go, delete nothing")
# Accepted after the subcommand too. Everyone writes the cron line as
# `prune-groups --config …`, and argparse only takes an option before the
# subcommand unless the subparser declares it as well. SUPPRESS so that
# leaving it out here does not overwrite a value given before it.
prune.add_argument("--config", type=Path, default=argparse.SUPPRESS)
prune.add_argument("--log-level", default=argparse.SUPPRESS,
choices=["DEBUG", "INFO", "WARNING", "ERROR"])
migrate = sub.add_parser(
"migrate", help="bring the database schema up to date")
migrate.add_argument("--config", type=Path, default=argparse.SUPPRESS)
migrate.add_argument("--log-level", default=argparse.SUPPRESS,
choices=["DEBUG", "INFO", "WARNING", "ERROR"])
args = parser.parse_args()
logging.basicConfig(
level=getattr(logging, args.log_level),
format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
)
cfg = load_config(args.config)
if args.command == "prune-groups":
sys.exit(asyncio.run(_prune_groups(cfg, args.days, args.dry_run)))
if args.command == "migrate":
sys.exit(_migrate(cfg))
single_worker_or_exit(cfg.server.workers)
uvicorn.run(
"meshbay_hub.app:create_app",
factory=True,
host=cfg.server.host,
port=cfg.server.port,
workers=cfg.server.workers,
log_level=args.log_level.lower(),
)
def migrations_dir() -> Path:
"""Where this installation's migrations actually are.
Derived from the package rather than written down, because the one place a
path like this can be correct is next to the code it describes. The RPM
installs `meshbay_hub` into a shared venv and copies `alembic.ini` to
`/opt/meshbay-hub/migrations/`, where `%(here)s/src/meshbay_hub/db/…`
resolves to a directory that does not exist — so the packaged unit's
`ExecStartPre` could never have succeeded, and a hub installed from the
package would not start at all. The same `%(here)s` trap had already been
found once on the server, with a copy of alembic.ini pointing at a
month-old snapshot of the tree.
"""
return Path(__file__).resolve().parent / "db" / "migrations"
def _migrate(cfg) -> int:
"""`alembic upgrade head`, with the paths resolved from the installation.
A command rather than a path in a unit file: it is correct for the RPM,
the DEB, a venv, and a checkout, and there is nothing to keep in step.
"""
from alembic import command
from alembic.config import Config
scripts = migrations_dir()
if not (scripts / "versions").is_dir():
print(f"meshbay-hub: no migrations at {scripts}", file=sys.stderr)
return 1
alembic_cfg = Config()
alembic_cfg.set_main_option("script_location", str(scripts))
# `env.py` reads the URL from the environment the same way the server does,
# so the two cannot drift; this is set for the case where it does not.
alembic_cfg.set_main_option("sqlalchemy.url", cfg.db.url)
command.upgrade(alembic_cfg, "head")
return 0
def single_worker_or_exit(workers: int) -> None:
"""Refuse to start with more than one worker.
Not a preference. `_connected_nodes`, `_node_groups`, `_webrtc_answers`
and the relay registry are per-process dictionaries: with two workers a
node registers in one and the WebRTC offers for it arrive at the other, so
the symptom is a node that is intermittently "offline" for half its
members. The mail allowance is in the database and would survive; nothing
else here would.
Refused at startup, where it is one line, rather than found later in a
report that describes something else entirely.
"""
if workers != 1:
print(f"meshbay-hub: server.workers is {workers}. This hub keeps its "
f"node registry and signaling state in memory and supports "
f"exactly one worker.", file=sys.stderr)
sys.exit(2)
async def _prune_groups(cfg, days: int | None, dry_run: bool) -> int:
"""Collect groups that were created and never given a node.
A group with no host has no files, no key and nothing to connect to, and is
invisible to everyone but its owner — so it is litter rather than data. The
grace period is counted from creation and `hosted_at` is never cleared, so a
node being offline today cannot make a live group look abandoned.
Deliberately a command rather than a loop inside the server: deleting other
people's groups on a timer nobody asked for is the kind of thing an operator
should schedule knowingly, and `--dry-run` lets them see the list first.
"""
from meshbay_hub.db.engine import close_db, get_session_factory, init_db
from meshbay_hub.tasks.cleanup import UNHOSTED_GRACE_DAYS, prune_unhosted_groups
grace = UNHOSTED_GRACE_DAYS if days is None else days
# init_db also runs create_all. That is what the server does at startup, so
# it adds no risk here — and it creates missing *tables* only, never a
# missing column, which is why deploys run alembic separately.
await init_db(cfg.db.url)
try:
async with get_session_factory()() as db:
gone = await prune_unhosted_groups(db, grace_days=grace, dry_run=dry_run)
finally:
await close_db()
verb = "would delete" if dry_run else "deleted"
if not gone:
print(f"No group older than {grace} day(s) is still unhosted.")
return 0
print(f"{verb} {len(gone)} group(s) unhosted for more than {grace} day(s):")
for gid, name in gone:
print(f" {gid} {name}")
return 0
if __name__ == "__main__":
main()
|