mirror of
https://github.com/PerpetualSoftware/pad.git
synced 2026-09-10 23:15:40 +00:00
xarmian
effd0199cd
* fix(events): detect a half-open Redis connection with a bus heartbeat (BUG-2738)
A Redis connection can stop carrying traffic without closing -- no FIN, no
RST, just a route that stopped working. The instance blocks on a read that
never returns, receives nothing, and its replay buffer goes on looking
complete, so every resume is answered "caught up" from a coverage window that
ended when the route did.
go-redis cannot see it: PubSub.Ping writes the command and never reads a
reply (v9.22.0), so its health check reports healthy for as long as the socket
accepts writes. Measured on day-52 against a proxy that silently stopped
forwarding: no reconnect in 24 seconds.
Each subscription now records when it last received ANYTHING -- event,
heartbeat, or subscription acknowledgement -- and a background pass ends the
coverage of any workspace whose stamp goes stale past 3T, then REPLACES the
connection. Drop alone would not recover: the resync it demands is served from
the same dead socket, so the detector fires again on the next pass.
Dave's day-49 ruling dissolves the threshold rather than tuning it. The bus
publishes its own frame every T=30s and fires at 3T=90s, which turns "is this
workspace quiet or is the route dead?" -- unanswerable, deployment-dependent --
into "did our heartbeat arrive?".
TWO PHASES, ORDER NOT OPTIONAL. The frame must travel on the workspace's event
channel, because that connection is what needs proving. A pre-phase-1 binary
cannot classify it: the frame reaches the event decoder, fails, and since
BUG-2739 that is a hole in coverage -- so an early flip makes every un-upgraded
instance drop its buffer and resync all its clients, every 30s, per workspace,
for the length of a mixed deployment. Phase 1 recognises and ignores;
PAD_EVENTS_HEARTBEAT is phase 2, a constructor parameter with no default so
every call site states its phase.
The idle detector is a THIRD actor in a region whose invariants were designed
around request goroutines plus Close. Four rules, each commented at
cycleIdleSubscriptions and each with a test:
1. It refuses to cycle while pendingSubs holds a record, and MINTS the
record itself before tearing anything down -- subscribeAndReplay checks
pendingSubs before wsSubs, so a subscriber arriving mid-cycle joins the
replacement instead of being admitted into the doomed subscription.
2. lastSeen is stamped at INSTALL, not left at the zero value, which reads
as 1970 and would cycle hardest on an unconfirmed admission -- the
workspaces already having a bad time.
3. wsCounts is re-read under the lock that performs the teardown.
4. Re-establishment runs on b.ctx with a nil establisher; the bus has no
subscriber registration of its own to unwind.
Two decisions beyond the plan:
A NEW COUNTER, not just the reset reason. dropWorkspaceCoverage reports a
reset only when a buffer existed to drop, and the incidents this detector
exists for skew hard toward having none -- a route that wedged early on a
quiet workspace. Reading cycles off the reset label alone would under-report
exactly the case it was built to find, so pad_event_subscription_cycled_total
is the dependable count and idle_timeout is corroboration. Both comments say
which is which.
THE CADENCE IS A LIVE TUNABLE -- a timer re-read under b.mu each pass plus a
buffered kick, not a ticker constructed once. A ticker captures the interval
at goroutine start, which makes the field write-once while its comment calls
it a tunable and makes any later write a data race; it also leaves no
deterministic way to test the WIRING other than a test-only constructor.
decodePayload's signature grew a payloadKind. The classification belongs to
the decoder, not the call site, so no future caller can reintroduce the
coverage drop; and the prefix (rather than an exact payload) means a later
frame version needs no third roll.
Also swept, per the team's prose convention: receiveMessages' doc comment and
deployment.md both said this gap was open and needed a decision. Both now say
what closes it -- and deployment.md says the watch stream still has the same
defect by the same mechanism, which is its own unit.
Trio kept together: ResetReasonIdleTimeout, the metric Help strings, and
docs/deployment.md's rollout order with the mixed-fleet failure named.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* test(events): rebuild the instruments the BUG-2738 matrix showed were blind
The mutation matrix found a defect in the fix itself and three tests that
could not have caught what they were named for.
THE DEFECT: the idle scan skipped a subscription whose lastSeen was the zero
value. That reads as belt-and-braces beside the install-time stamp and is the
opposite -- it makes a subscription that has NEVER received anything
permanently uncyclable, which is the BUG-2747 unconfirmed admission: the one
population the plan singles out as mattering most, and the one where a wedged
route would then be undetectable forever. It was also masking rule 2: with the
skip present, removing the install stamp survived every test. Skip removed;
that mutation is now caught. Re-adding it is undetectable by construction and
the comment says so, because a guard that only acts once a real one has broken
converts a caught defect into a silent one.
THREE INSTRUMENTS THAT WERE NOT MEASURING:
- "Drop only, never cycle" passed because establishSubscription overwrites
wsSubs, so a generation check cannot see a replacement installed WITHOUT
tearing the old connection down -- a leaked PubSub, connection and receive
goroutine per cycle, forever, on exactly the wedged route where they never
die on their own. Now asserted on the receive loop exiting.
- The Close test was vacuous. Close drains wsSubs, so a loop that ignored
b.ctx entirely would find no workspaces and publish nothing: silence after
Close was evidence of nothing. maintenanceStopped makes the goroutine's exit
observable, which is the same reason Observer.ReceiveLoopExited exists.
- The joint test HUNG rather than failing under the drop-only mutation: the
seam never fires, so the joiner goroutine was never spawned and an unbounded
receive waited forever. The harness then aborted mid-run and LEFT THE
MUTATION APPLIED to the working tree, which a grep caught and a green test
run would not have. The wait is bounded and names the failure; the harness
bounds each run, reports a hang as its own status, and restores in a finally.
Added: a direct test that a straggler frame from a replaced generation cannot
refresh its successor's liveness -- on a wedged route, the dead connection's
buffered tail would otherwise suppress the detector for the replacement.
RULE 3 IS AN OPTIMISATION, NOT A CORRECTNESS GUARD, and the matrix says so
rather than an argument: removing the whole second read -- liveness, generation
and count terms together -- survives every test, because
establishSubscription's abandon path already refuses to install for an emptied
workspace and retires the record in the same critical section (BUG-2749). The
first read is redundant more sharply still: reaching zero takes the
subscription down with it, so this loop never sees such a workspace. Both are
kept, because neither DEPENDS on that coupling, and both comments now carry the
per-term reading instead of describing tested defence in depth. The generation
term is unreachable while the establishment record is held, by rule 1's own
mechanism.
Matrix: 16/22 detected, plus 4 follow-ups. Every survivor is documented at its
line with why it survives.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* fix(events): gate idle detection on heartbeat phase 2 (BUG-2738, codex r1)
Codex round 1 found a defect the first draft had shipped WITH A COMMENT
JUSTIFYING IT, plus two coupling hazards.
P2-as-filed, P1 in effect: idle detection ran on every instance from phase 1,
on the reasoning that it could "detect off whatever traffic the deployment
already carries". That holds only for a BUSY workspace. A QUIET one on phase 1
has no events and no heartbeat, so a perfectly healthy subscription crossed
the 90s threshold on every pass and was cycled: replay coverage dropped, every
live subscriber told to resync, indefinitely -- on the DEFAULT configuration
every deployment lands in before it flips anything. A resync storm shipped as
the default, by the feature whose stated purpose is to avoid exactly that load
inversion.
Publishing and detecting are now one switch, which is what they always were:
an instance detects off its OWN frames -- it publishes to the channels it
subscribes to and receives them back -- so it never depended on peers having
flipped, and there was never a reason for the two to be separable. Phase 1 is
"recognise the frame so a phase-2 peer costs you nothing", and nothing else.
Regression test plus its counterfactual, so "no cycles" cannot be satisfied by
a detector that has simply stopped working.
P1: the maintenance loop published heartbeats and scanned for idleness on one
goroutine. publishHeartbeats makes N synchronous Redis publishes, and against
the failure this feature exists to detect those are precisely the calls that
block -- bounded by go-redis's own Dial/Read/WriteTimeout, not by any context
we can pass. A stalled publisher could therefore delay detection for as long
as those timeouts take, on the very instance whose connections had wedged, and
for longer the more workspaces it carried. Two goroutines with their own kick
channels; a stalled publisher now just produces silence, which is what the
detector reads.
P3: the cycle held the workspace's establishment record across a synchronous
observer report, so an Observer callback that subscribed to that workspace
would wait on a record only the reporting goroutine could retire. Moved the
SubscriptionCycled report past establishment. The narrower half is older than
this code -- confirmSubscription's late-acknowledgement path already reported
from inside that window -- so it is documented on the Observer interface as a
contract rather than silently worked around: a callback may publish, read and
unsubscribe; it may not subscribe.
Prose swept for what the gate falsified, per the team convention: the
constructor comment that argued for the defect, config.EventsHeartbeat's
rollback paragraph, the config test's inverted-rationale comment,
ResetReasonIdleTimeout, both metric Help strings, and deployment.md's phase
table and rollback section. All of them now say that phase 1 detects nothing
and that the cycled counter is STRUCTURALLY zero there -- a zero on phase 1
says nothing about whether a route has wedged, which is the reading an
operator would otherwise get wrong.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* test(events): prove a resuming joiner is told sync_required across a cycle
Codex round 2 raised that a subscriber arriving DURING an idle cycle gets no
gap signal, because dropWorkspaceCoverage only signals subscribers present
when it runs. True, and for a RESUMING caller the gap signal is not what
protects it: the registration mark is. It registers while the workspace has no
buffer, so its mark cannot match whatever buffer exists by the time it reads,
and eventsSinceMarkLocked answers nil -- sync_required rather than a false
"caught up".
A FRESH caller is deliberately not signalled and the finding is DECLINED for
that case, with reasons recorded at the test: it holds no prior position, so
there is no span it could be missing; it is admitted only after the
replacement subscription is acknowledged, because it waits on the cycle's
establishment record which finishPending closes after the confirmation; and on
the unconfirmed-admission path it IS told to reconcile when the acknowledgement
lands. Signalling it anyway would demand a resync of a client with nothing to
reconcile -- the load inversion this unit already had to fix once.
THE FIRST TWO VERSIONS OF THIS TEST DID NOT DISCRIMINATE, which is the part
worth keeping. Version one asserted the empty case: the cycle leaves no buffer,
so eventsSinceMarkLocked returned nil from its `!ok` term and removing the mark
check entirely still passed. Version two published inside
afterSubscriptionConfirmed so a FRESH buffer exists before the joiner reads --
and deleting the `mark.buffer == nil` term still survived, because the keep
arithmetic in that function already reduces to zero for a nil mark. Only
replacing eventsSinceMarkLocked with the unmarked eventsSinceLocked fails the
test, handing the joiner the post-cycle event as though it followed its cursor.
That is the mutation the test is built against, and the redundancy inside
eventsSinceMarkLocked is recorded rather than mistaken for coverage.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* fix(events): only count a cycle that actually replaced the connection (codex r3)
Three findings from a fresh-angle round on shutdown, wire format and doc
accuracy. The wire-format angle came back clean -- events:<workspace> cannot
collide with watchevents under validated namespaces, and no valid activity
payload can be mistaken for an hb| frame.
P3, and the one that stings: config.EventsHeartbeat still said phase 1
"already runs idle detection off whatever traffic exists". That is the exact
sentence the previous commit's sweep existed to remove, in a file that sweep
edited. A grep for the phrasing I remembered writing missed the paraphrase
sitting four lines above the paragraph I did fix.
P3: SubscriptionCycled was reported unconditionally after establishSubscription
returned, but establishment has two reasons to install nothing -- the bus
closed, or the workspace emptied while we dialled. The counter's documented
meaning is "torn down AND replaced", and counting an aborted establishment is
wrong in the direction that matters: an operator reading a non-zero rate
concludes connections are being blackholed, so a shutdown would manufacture
that signal. Now reported only when a replacement is installed, verified by
generation. Both Help strings and deployment.md say "counts replacements, not
teardowns"; the teardown stays visible through the idle_timeout reset reason.
P2: Close does not join the maintenance goroutines. Kept that way and
documented on Close, because the publish half makes synchronous Redis calls
bounded by go-redis's own timeouts -- the calls that stall on exactly the
wedged route this feature detects -- so joining would let a dead network hold
shutdown open. What has to hold instead is that a cycle already past its ctx
check leaves nothing behind, which is now pinned by a test that closes the bus
from inside the cycle's establishment: no subscription installed, no
establishment record stranded, no counter moved.
liveGen moved from the test file into the package -- production needs it now.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* test(events): restore the coverage the phase gate silently removed
The mutation matrix, re-run against the post-codex code, showed M3 -- removing
the install-time lastSeen stamp -- going from DETECTED back to SURVIVED. The
cause was my own round-1 fix: gating idle detection on heartbeat phase 2 means
a phase-1 bus never scans, and TestAnUnconfirmedAdmissionIsNotCycledAsIdle
built its own phase-1 bus. It was the only test that could observe a zero
lastSeen, because the plain fresh-subscription case is stamped twice over --
at install, and again by the acknowledgement. Flipped to phase 2 and
re-verified: removing the stamp fails it again.
Worth naming the shape rather than just the fix. A behaviour change that
narrows when code runs silently narrows what the tests reach, and nothing in a
green suite says so -- the tests still pass, they just stopped asking. Only
re-running the matrix after the change surfaced it.
Two harness bugs fixed alongside, both of which had been reporting
non-results as if they were readings:
- A mutation that INSERTS keeps its own anchor, so the "did the edit land?"
check read every insertion as ANCHOR-ERROR. It compares the file now.
- The two rule-3 mutations left `sub`/`live` unused and came back BUILD-BREAK
rather than answering the question; they carry the same discard the
follow-up harness already used.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* test(events): close the wiring and barrier gaps codex round 4 found
Concurrency and lock discipline came back CLEAN -- the establishment record
and the generation checks cover two racing cycles, Unsubscribe, Publish and a
stale resubscription frame, with no lock-order deadlock. The four findings
were all about whether the tests measure what they claim.
P2, and it is the convention I had cited three commits earlier: the heartbeat
flip had no wiring test. internal/events proves a bus built with
publishHeartbeat=true emits frames and detects idleness, and every one of
those tests passes if newObservedEventBus hardcodes false -- the deployment
would simply never detect a wedged connection, which is indistinguishable from
a deployment that has none. Both directions asserted, because a helper that
ignored its config and hardcoded EITHER value passes a one-directional test.
Mutation-checked against exactly that edit.
P2: the metrics adapter test never touched SubscriptionCycled or the
idle_timeout reason, so an adapter that folded the counter into the reset
series -- destroying the very distinction those two are built to keep apart --
would have passed. Both added with counts that differ from their neighbours',
the pattern that file already uses so a label-dropping adapter cannot satisfy
the totals by coincidence.
P3: TestAHeartbeatConsumesNoEventID "waited" on a predicate that returned true
unconditionally. Not a slow wait -- no wait at all: the counter was read with
the publishes still in flight, so a heartbeat that DID consume an id could
land afterwards and the test would still pass. It now waits on the frames
arriving, and fails against a mutation that publishes an event alongside each
heartbeat.
P3: the maintenance goroutines started on phase 1, where both halves are
guaranteed no-ops -- two goroutines and two timers per process waking every
30s for the life of a deployment that asked for none of it, and phase 1 is the
DEFAULT. The flag is constructor-only so the decision is taken once. The
in-function gates stay: those are the correctness ones, and the tests reach
them directly without a loop.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* fix(events): validate the heartbeat frame and stop serialising recovery (r5)
Client-facing behaviour came back CLEAN: an idle cycle signals each local
subscriber, the SSE handler emits an in-band sync_required with an empty id
while holding the connection open, EventSource retires its cursor and the web
client runs the documented reconciliation. Two P2s on the other angles.
FRAME VALIDATION. Accepting any "hb|..." created a silently-ignored class on
the workspace event channel, where before this feature EVERY unreadable
payload ended coverage loudly and moved undecodable_message -- the counter
whose documented job is "suspect a namespace collision". A foreign or buggy
publisher whose bytes happened to start with the prefix slipped through that
signal without a trace. A frame is now hb|<version> plus optional short tokens
under a length cap; anything else wearing the prefix goes back to being a
coverage-ending decode failure, and the forward compatibility the prefix was
chosen for survives for a disciplined future frame.
What this deliberately does NOT try to fix, because it is not a hole: a forged
frame cannot fake liveness. Liveness means "this socket carried traffic", and a
frame that ARRIVES demonstrates exactly that whoever sent it -- which is why
stampLastSeen already fires for undecodable frames. There is no coverage claim
inside a heartbeat to forge.
CADENCE DRIFT, which was self-defeating rather than merely untidy. The timer
restarted after each pass, so the real period was T plus however long the pass
took. For the publisher that means an instance whose publishes are slow emits
heartbeats further apart, its own subscription sees them further apart, and it
can cross its own 3T threshold and cycle connections that were never wedged --
the slowness manufacturing the incident. Scheduling is deadline-based now, and
resets rather than bursting when a pass overruns badly.
SERIAL RECOVERY. One idle pass re-established every due workspace in sequence,
each re-dial bounded by go-redis's own timeouts, so recovery took N x that
timeout with the last workspaces reporting themselves uncovered throughout.
The failure that puts many workspaces on the due list at once is a Redis
failover, so the serial case was the common one. Bounded-parallel at 8 -- each
entry already owns its establishment record so they are independent by
construction, and an unbounded fan-out would answer a struggling Redis with one
dial per workspace at once. Test covers more workspaces than the cap, and
fails against a version that drops the overflow.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* docs(events): idle_timeout means coverage ended, not connection replaced (r6)
Codex round 6 came back clean on the non-Redis path (MemoryBus ignores the
Redis-only flag; EventBus and Close have not drifted), on the rollback
rehearsal (phase-2 to phase-1 and a mixed fleet are safe as documented,
including a bus mid-cycle -- Close cancels it, prevents installation and
retires its pending record), and on the operator surface
(PAD_EVENTS_HEARTBEAT is a server env/TOML setting; `pad configure` is client
connection config and needs no new surface).
The one finding is a contract drift I introduced two commits ago and then
wrote prose for in the same commit. Making SubscriptionCycled mean "replaced"
was right; what I missed is that the idle_timeout RESET REASON is emitted
earlier -- dropWorkspaceCoverage runs before the re-establishment -- so it can
fire when nothing is replaced, which is exactly the shutdown case the counter
was changed to exclude. Three doc sites and one log line said "replaced the
connection" anyway.
They now say what is true at the moment each fires: idle_timeout means
COVERAGE ENDED, only pad_event_subscription_cycled_total proves a replacement,
and the log says "attempting to replace" rather than "replacing". The log
wording matters on its own -- an operator correlating it with the counter
would otherwise find the log without the counter and go hunting a bug that
isn't there.
Third time this unit has produced prose the next change falsified, and each
time a different reviewer angle caught it rather than the sweep I ran at the
time. The pattern is that a behaviour change and the prose describing it land
in one commit, so there is no diff between them to notice.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* test(cmd): drive the heartbeat wiring test instead of sleeping at it (r7)
Codex round 7 found no leftovers across seven rounds of edits, and confirmed
the mass-cycle case does NOT produce a reconnect storm -- the SSE connections
stay open across a sync_required, so the admission limits are never consulted.
P3, and it is the failure I have been criticising in other people's tests: the
wiring test used a 300ms sleep as its ordering barrier. Under -race or on a
loaded CI box, a phase-1 bus that is correctly silent and a phase-2 goroutine
that merely has not been scheduled yet are indistinguishable, so the test could
pass or fail for reasons unrelated to the flip it exists to check. It now
drives one publish pass synchronously through a named test hook and uses an
ordinary event on the same channel as the barrier, which Redis delivers in
publish order. No timing left. Verified: still fails against the flag being
hardcoded false, and ten consecutive -race runs are green.
That replaces SetMaintenanceCadenceForTest with PublishHeartbeatsForTest rather
than adding to the exported test surface -- the loop's own wiring is covered
inside internal/events, where the unexported setter is available.
P2 is FILED, NOT FIXED, as BUG-2761: a mass coverage drop tells every connected
subscriber of every affected workspace to resync at once, and each browser tab
independently calls /changes with per-tab coalescing but no jitter and no
global budget. The fix is a web-client change plus possibly a wire-format hint,
which is independent of half-open detection and would materially expand this
diff. Worth filing rather than shrugging at because this unit makes the
simultaneous case MORE likely: it adds a third trigger of a class that already
existed (Redis failover, epoch change), and its natural cause is exactly a
network event that wedges many routes at once. deployment.md carries the
residual with the bug ref so an operator meets it before the incident does.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* test(events): make the tests prove what their comments claim (codex r8)
Round 8 was claim verification rather than bug hunting -- check the diff's
load-bearing assertions against the actual code -- and it was the highest-yield
round of the eight. The go-redis assertions (Ping writes without reading, the
channel path sets no read deadline, TLS dials ignore cancellation) and the four
claims about neighbouring functions all held. Seven other assertions did not.
TESTS THAT DID NOT PROVE THEIR OWN HEADLINE. This is the substance of the
round, and every one of these passed before and after:
- The JOINT TEST -- this unit's flagship -- claimed to discriminate the
two-subscriptions failure and did not. Fan-out is per subscriber, so a joiner
that opened its OWN second subscription still delivers the event to everyone
exactly as the test expected. Nothing separates one subscription from two
except counting them, which it now does at Redis, plus a duplicate-delivery
check for the second receive loop. Fails against the pending record not being
minted in the scan.
- The remedy test said "the old connection must also be gone" and waited for a
receive-loop exit. stopRedisSubscription does two things and the loop exits on
the first alone, so it passed against a version that cancelled the loop and
left the PubSub and its health check open. Counted at Redis now; fails against
exactly that mutation.
- The parallel-recovery test could not tell serial from parallel -- a serial
pass cycles all thirteen workspaces too. It now uses a rendezvous, asserts the
peak concurrency is above one AND within the cap, and fails against a serial
implementation.
- The prefixed-garbage test only exercised the classifier. Whether
receiveMessages ACTS on the error is a different claim, now driven through
the real Redis path.
- The metrics adapter test's comment said "every reason this bus can emit"
while subscription_unconfirmed was missing; its zero-assertion proved
non-leakage, not mapping. Emitted now with a count distinct from its
neighbour's, so a merging adapter cannot satisfy both.
PROSE THAT OUTLIVED THE CODE, again. The latency arithmetic still described the
single shared ticker that round 5 replaced with two independent loops; from
lastSeen [3T,4T) still holds, but from FAULT ONSET it is roughly [2T,4T)
because the publisher has its own phase. And a second "and replaces the
connection" in deployment.md that round 6's sweep missed.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* docs(events): correct three contract statements (codex r9)
Round 9 was cross-artifact conformance: every commitment the plan made was
checked against the code. All met -- wire classifier, lastSeen placement and
locking and install stamp and every-frame stamping, heartbeats bypassing
Publish and the shared counter, the drop-and-cycle remedy under the
single-establisher invariant, all four joint rules, the two-phase rollout with
its inverted-rationale test, and the reason/Help/deployment.md trio with the
rollout order. It also confirmed the three documented mutation survivors are
correctly dispositioned: both wsCounts checks are redundant-but-cheap under the
current invariant, and omitting the lastSeen.IsZero() skip is right because
adding it would mask a regression in the install stamp.
Three statements were wrong.
The env-var contract. My test comment said an unparseable PAD_EVENTS_HEARTBEAT
"must leave the flip off", which is true from a default config and false from a
config file that set it true -- there the value is left alone, as the
precedence test already asserts. The BEHAVIOUR is right and matches the epoch
flag: a typo must not move a migration in either direction, and silently
rolling an operator back to phase 1 would disable detection on a fleet that had
opted in with nothing saying so. Only the prose overclaimed, and it overclaimed
in the direction that invites someone to "fix" the ignore into a fail-closed
reset.
The constructor. NewRedisBusWithKeys documented publishEpoch and said nothing
about publishHeartbeat sitting next to it -- two adjacent booleans of the same
type belonging to two independent migrations, which is a shape that gets
swapped or dropped in a maintenance edit. Both now documented in order, with a
note that any combination is valid.
A stale count. EventSequenceResetsTotal's comment said "Five reasons" and there
are seven; it was already wrong by one before this unit added another. Replaced
with the count plus a pointer to the three artifacts that are authoritative and
move together, since the count itself is the part that goes stale first and is
read last.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* test(events): make the cadence arithmetic testable, and justify a guard pair
Matrix 5 (29 mutations, 21 detected) surfaced two things the previous run
could not, because both concern code the codex rounds added.
THE DRIFT FIX HAD NO TEST. Restoring the sleep-after-work form survived every
test in the package, and would have kept surviving: the only way to observe
drift through the loop is to time it, and a timing assertion is a flaky
assertion. Extracting nextTick makes the arithmetic checkable without a clock,
and the four cases now pin what the schedule is for -- a slow pass does not
push the next tick out, ten slow passes accumulate no drift, an overrun beyond
one interval resets instead of replaying the missed ticks, and an overrun
WITHIN one interval still catches up rather than re-phasing the schedule
permanently. Both directions mutation-checked.
The property is worth this much because breaking it is self-defeating rather
than merely untidy: an instance whose passes are slow emits heartbeats further
apart, its own subscription sees them further apart, and it crosses its own 3T
threshold and cycles connections that were never wedged.
A GUARD PAIR THAT ONLY DIES TOGETHER, which the team lesson says to treat as a
question rather than a clearance. The loop's ctx.Done select arm and its
post-wait ctx check each survive removal alone. Checked rather than assumed:
they cover disjoint moments and each is independently right -- the select arm
is the exit while WAITING, which is where the goroutine spends its life, and
the post-wait check stops a bus that closed DURING a pass from starting
another one against a cancelled context and a drained wsSubs. Removing BOTH is
detected. Reasoning recorded at the code, and the combined mutation added to
the matrix so the pair cannot quietly become a single point of failure.
Also fixed an ineffassign the lint gate caught in the new test.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* docs(events): state what the detector does not cover (codex r10)
Round 10 was adversarial: refute the unit's central claim rather than look for
defects in it. It partly succeeded, and the corrections are worth more than
most of the bug findings.
The claim was "a wedged connection is detected, coverage is ended, and the
connection is replaced so delivery resumes". Three parts of that were too
strong, and all three limits were checked against go-redis v9.22.0 rather than
argued:
IT IS A RECEIVE-SIDE DETECTOR, not a round-trip health check. It measures
whether frames ARRIVE. A subscription whose outbound direction is broken but
which still receives reads as healthy -- correctly, since nothing is lost, but
that is a narrower claim than "the connection is healthy".
IT CANNOT COVER THE PUBLISH PATH. PUBLISH travels on the client's connPool
while a subscription holds a connection from the separate pubSubPool
(redis.go:363, :1956) -- different sockets, different fates, and a reconnect of
one repairs nothing about the other. An instance whose publish path is wedged
loses its own events for every other instance and this feature will not say so.
That is a real gap in the family's coverage, now written down rather than
implied away.
REPLACEMENT IS ATTEMPTED, NOT GUARANTEED. If the path is still blackholed when
the cycle re-dials, the replacement cannot receive either. Coverage stays ended
so nothing is falsely claimed, but delivery resuming is a statement about the
network rather than about this code.
Filed BUG-2764 rather than folded in: establishSubscription's
`b.client.Subscribe(dialCtx, channel)` silently discards the SUBSCRIBE error,
because go-redis's own Client.Subscribe drops it (`_ = pubsub.Subscribe(...)`,
redis.go). A failed subscribe therefore installs a connection that looks live
and is subscribed to nothing. It is pre-existing, it lives in the establishment
path three bugs have already converged on, and changing how that function
issues its SUBSCRIBE does not belong in a diff about idle detection. Worth
knowing here because it is the one way the replacement can fail on a HEALTHY
network -- and because the detector now cycles it on the next pass, which is
why it self-heals on phase 2 and stays dead forever on phase 1.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* fix(events): do not cycle a workspace that recovered before its turn (r11 P1)
Codex round 11 attacked three claims. Phase-1 safety and rollback safety both
came back clean -- a phase-1 receiver stamps lastSeen and nothing else, touches
no buffer, metric, client, ID or epoch, and its maintenance loop is not started
at all, so that timestamp is inert; heartbeats leave no state in Redis or
across a process replacement, and a mid-cycle shutdown rechecks b.ctx before
installing. The third claim did not survive.
FALSE POSITIVES ON A HEALTHY SYSTEM, which is the property this design cares
about most: cycling a working subscription drops its coverage and resyncs every
one of its subscribers for nothing.
cycleIdleSubscriptions selects its victims under the lock and releases it; the
cycles run afterwards. Its re-checks asked about generation, subscriber count
and bus liveness -- and never re-asked the question the scan had asked. A
subscription that started receiving again in that window was cycled anyway.
The window is not theoretical, and this unit widened it itself: the 8-way
concurrency cap added in round 5 makes a workspace wait behind earlier batches
of slow replacement dials, and a GC or CPU pause leaves a backlog of heartbeats
undrained in the receive loop. Both are ordinary conditions on a loaded box.
cycleOne now validates, ends coverage and tears down WITHOUT RELEASING THE LOCK
in between, which needed dropWorkspaceCoverage split into a locked variant that
returns its reason for the caller to report after unlocking. That also removes
the ordering fragility the previous version documented rather than fixed: there
is no longer any window in which coverage is ended for a workspace this
function then decides to leave alone. The log moved after the decision for the
same reason -- it could previously describe a cycle that then abandoned.
The freshness term is load-bearing and says so, next to the three neighbouring
terms whose mutation survivals are recorded as redundant-but-cheap. Removing it
is detected, by a test that lands the recovery in the exact gap through a new
positional seam.
NTP steps were checked and are not a hazard: time.Time carries a monotonic
reading, so a wall-clock step cannot make a subscription look idle.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* perf(events): take logging and PubSub.Close off the global lock (codex r12)
Round 12 verified round 11's freshness fix: validation, coverage invalidation
and teardown are atomic under b.mu with no lock cycle,
dropWorkspaceCoverageLocked preserved the original semantics exactly including
the no-buffer branch that still signals subscribers, reset reporting happens
after unlocking, and the replacement metric still lands only when a new
generation does. Slow establishment stays outside b.mu, wg.Wait only delays the
next pass, and Close cancellation retires pending records.
Two P2s, both about what round 11 put UNDER that lock:
slog.Warn ran while b.mu was held. slog invokes the installed handler
synchronously, and b.mu is the lock every fan-out and every Subscribe on the
instance contends for -- a slow or custom handler stalls all of them, and one
that calls back into the bus deadlocks. Moved after the unlock; it still has to
come after the DECISION, for round 6's reason, so both constraints are now
stated together at the call.
PubSub.Close ran under b.mu too. It takes go-redis's own mutex, which the
health check can hold across reconnect work, so a network-bound wait sat inside
the instance's hottest lock. That was survivable when teardown only happened as
a workspace lost its last subscriber; the idle detector makes it happen on
every cycle, which is what turned a latent cost into a real one. Handed off to
a goroutine: nothing references the PubSub once the map entry is gone, and
cancel() -- which is what actually stops delivery -- still happens under the
lock.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* fix(events): do not read our own failed probe as a dead peer (codex r13)
Round 13 asked for a production-approval review. Four findings; the second is
the sharpest of the whole run because it is the mirror image of the failure
this feature exists to find.
A FAILED HEARTBEAT PUBLISH WAS READ AS A DEAD SUBSCRIPTION. The detector's
inference is "we published a frame and nothing came back, so the receive path
is dead" -- valid only if the publish actually happened. PUBLISH travels on the
client's connPool while the subscription holds a connection from the separate
pubSubPool, so a publish-side failure (pool exhaustion, a wedged outbound
route, Redis refusing writes) says nothing about whether that subscription can
receive. The detector was reading its own inability to probe as evidence about
the peer, and tearing down healthy connections on a schedule: a resync for
every subscriber of every workspace, every 90s, for as long as the outbound
path stayed broken. The third load inversion this unit has had to fix.
redisSub.lastProbeOK now records the last SUCCESSFUL publish, and detection is
suspended while it is stale -- checked in the scan and again in cycleOne, which
is a pair that only dies together and is therefore justified at the code:
the scan's keeps a workspace off the due list so no record is minted and no
joiner waits, cycleOne's covers the probe failing AFTER selection, a window the
concurrency cap makes real. Neither subsumes the other; removing both is
detected. New counter pad_event_heartbeat_publish_failures_total, documented as
DETECTION DEGRADED rather than as a peer being broken.
THE END-TO-END TEST THAT DID NOT EXIST. Every other test drives this through a
fake clock -- necessary, since the threshold is 90s by construction and
miniredis always answers, but it means they all ASSUME the wedge rather than
produce it. A TCP proxy that stops delivering server->client on the connections
already open, while writes keep succeeding and new connections stay healthy,
produces the real thing. The test asserts both halves of the claim: the wedge
is detected, and the replacement delivers. Both halves mutation-checked
(detector disabled; drop-only with no replacement).
The proxy's first version was vacuous -- a global flag consulted at read time
meant re-enabling delivery for future connections also revived the ones meant
to be dark. Per-connection now, and the comment says why.
Also: PubSub.Close taken off b.mu in Close() too (round 12 fixed only the cycle
path), and the replacement counter now takes an explicit installed result from
establishSubscription rather than inferring one from the live generation --
inference misattributed an unrelated caller's fresh subscription as this
cycle's replacement, and missed a real replacement that had lost its last
subscriber. Both mutation-checked.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* fix(events): bind the probe stamp to a generation; make the proxy test honest
Round 14 returned a BLOCK verdict on two P2s, both mine, both in the fix that
round 13 had just added.
lastProbeOK WAS NOT GENERATION-BOUND. publishHeartbeats snapshots the workspace
list, publishes off the lock -- for as long as go-redis's timeouts allow -- and
then stamped whatever subscription occupied that workspace by the time it
returned. A probe sent for generation A could credit generation B, which never
received one; if later probes then failed, B could be cycled while looking
recently probed. Exactly the hazard stampLastSeen already guards on the same
map, and I did not carry it across. The generation now travels with the
snapshot and is validated before stamping.
THE END-TO-END TEST COULD PASS WITHOUT EXERCISING WHAT IT CLAIMED. It darkened
the receive direction of every open connection, including the ordinary pooled
connection PUBLISH uses -- so the probe may have been failing too, and the run
would then have been exercising the cannot-probe path rather than a half-open
route, which is the very distinction round 13 added the premise check for. The
proxy now classifies connections as it forwards and darkens only one that has
carried a SUBSCRIBE, leaving the publish path healthy, and the test asserts
zero probe failures so a run that drifts back into the other case fails loudly
instead of passing quietly. Still fails against a disabled detector and against
drop-only.
Also covered the new counter's mapping in the metrics adapter test, with a
count distinct from both neighbours -- cycled, idle_timeout and
heartbeat-publish-failure say three different things and an operator acts on
the difference.
Verified by the same round: install-time stamping does not permanently suppress
detection, establishSubscription returns false only on abandon and true on all
three installed paths including the cancelled-establisher goroutine, and
Close's deferred PubSub.Close runs after the unlock.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* test(events): pin the probe-across-replacement race (closes r15's residual)
Round 15 returned CLEAN and approve-with-comments, naming one residual: the
generation binding on lastProbeOK had no deterministic test, only the argument
that it mirrors stampLastSeen. This closes it with a positional seam between
the publish and the stamp, which is the only place that interleave can be
forced.
TWO INSTRUMENT DEFECTS ON THE WAY, both caught by mutation rather than by
reading:
The first version compared the credited stamp against the PROBE's timestamp.
On a frozen clock the replacement's install stamp and a wrongly-credited probe
are the same value, so it could not tell them apart -- it failed on the install
stamp while claiming a credit had happened, and removing the generation binding
still passed. It now compares against what the replacement was INSTALLED with,
and the clock advances inside the seam so a buggy write lands strictly later.
The second version was FLAKY: 2 failures in 3 runs. The heartbeat that was just
published comes back through miniredis on another goroutine, and if it lands
between the forced-stale write and the scan it refreshes lastSeen, the
workspace is not due, and no replacement happens. Retried until the generation
actually moves. Now 5 of 5 green unmutated and 5 of 5 detected mutated -- which
is the bar, because a 2-in-3 detector reads as coverage while being noise.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* fix(events): on-call signals — log the cycle outcome, correct two claims (r16)
Round 16 read the diff as the person paged at 3am. Four findings.
THE CYCLE LOGGED ITS ATTEMPT AND NEVER ITS OUTCOME. The line says "attempting
to replace", which is correct and, on the one path where the replacement does
not happen, left an on-call with a warning, no counter movement, and no
explanation. Now there is a second line naming the reason.
pad_event_receive_loop_exits_total's documentation was falsified by this unit
and neither doc site said so: every idle cycle stops a receive loop while its
subscribers are still connected, and the comment still claimed exits happen
only at shutdown or when the last subscriber leaves. Both sites corrected, with
the expectation that it tracks the cycle counter during an incident.
A CLAIM I MADE AND THEN COULD NOT SUPPORT, recorded rather than quietly kept.
Round 16 argued the age-based premise check ("has a probe succeeded within the
threshold") failed to suspend detection where an ordering rule ("has a probe
succeeded since anything last arrived") would, and I rewrote the rule on that
argument and wrote a test named for the defect. The mutation matrix then
refused to confirm it: reverting to the age form leaves the test green, and so
does removing both copies of the check, and no case separates the two — on any
healthy path the two stamps advance together, because a probe whose frame
arrives sets both, and they diverge only on the wedge where both forms cycle.
The ordering rule is kept, because it states the intent exactly and is never
weaker. But the test and the comment now say what they actually establish —
that a probe which has started failing stops the detector concluding from
silence, which is the property both forms share and neither had before — rather
than claiming a fixed defect I cannot demonstrate.
The two remaining P2s are already-filed residuals: the cycled counter proves an
install rather than a working replacement (BUG-2764), and repeated cycling
amplifies /changes load with no jitter or global budget (BUG-2761). Both are
documented in deployment.md with their refs.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* docs(events): record what the final matrix actually says about four guards
Final matrix: 34 mutations, 22 detected, baseline restored green. Every
survivor is now documented at its line with why it survives, and two of them
turned out to be instrument defects rather than coverage gaps.
lastProbeOK's INSTALL STAMP IS REDUNDANT and the comment claimed otherwise. It
said a zero value "would permanently disqualify a subscription from ever being
cycled" -- true of the age-based premise it was written for, false under the
ordering rule that replaced it, because a zero value fails
`lastProbeOK.After(lastSeen)` exactly as an install stamp equal to lastSeen
does. Kept, for a reason it earns: it makes the field's invariant true by
construction, so a future rule reasoning about this value's AGE gets a real
timestamp rather than 1970 -- which is the trap the age-based rule fell into
one field over.
THE TWO cycleOne ABANDON GUARDS DIE ONLY TOGETHER AND ARE NOT REDUNDANT, which
took checking rather than assuming. They catch different shapes of the same
recovery: an arrival that has not been re-probed pushes lastSeen past
lastProbeOK so the premise case fires and the freshness case is unreachable --
that is the shape the test produces, and it is why removing either alone stays
green. But the publisher runs on its own goroutine at its own cadence and can
land a successful probe between the arrival and the decision, putting
lastProbeOK ahead again; there only the freshness case stops a healthy
subscription being torn down. Deleting it on the strength of the matrix would
remove the second shape's only guard.
Close's off-the-lock PubSub.Close is UNTESTED BY DESIGN, recorded rather than
papered over. It is a contention property, and the only assertion that
separates it is a timing one, which in this suite is a flaky one.
TWO HARNESS DEFECTS, both of which produced false survivors that would have
gone into the evidence package as findings. M11a inserted its mutation AFTER
the gate it was meant to disable -- unique anchor, wrong placement, so the
early return still fired and nothing changed; with a correct anchor it is
detected. M20 left variables unused and came back BUILD-BREAK rather than
answering; in compiling form it genuinely survives, consistent with
establishSubscription's abandon path already covering it.
The lesson worth keeping: when I rewrote all 34 anchors against current source
I verified each matched exactly ONCE, and uniqueness is not placement. An
anchor can be unique and still land somewhere that changes no behaviour.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
* test(events): barrier the probe test on delivery — it was flaky, CI caught it
Go (PostgreSQL) failed on af7001ab, in a test I added two commits ago. Not a
timeout and not the race step: TestAFailedProbeAfterASuccessfulOneStillSuspends
Detection asserted no cycle and got one.
The test killed Redis immediately after a successful probe, without waiting for
that probe's frame to be delivered back. If the frame never lands, lastSeen
stays at the install stamp, the successful probe is then legitimately "after
the last arrival", the workspace is genuinely due — and the code cycles it FOR
THE RIGHT REASON under a test asserting it should not. The premise the test is
named for simply did not hold on a slower machine.
So this was not a false alarm in CI and not a defect in the code: it was my
test asserting an outcome whose precondition it never established. Waiting for
lastSeen to move makes the precondition real. Eight consecutive local runs
green, and removing both premise checks still fails it, so the barrier did not
neuter what it was measuring.
Worth naming because it is the third instrument defect in this unit found by
something other than reading it — after the harness restore that ate an edit
and the unique-but-misplaced mutation anchor. A test that depends on an
unsynchronised delivery is a test that passes on the machine that wrote it.
Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X
1369 lines
59 KiB
Go
1369 lines
59 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"io/fs"
|
|
"log/slog"
|
|
"os"
|
|
"os/exec"
|
|
"os/signal"
|
|
"path/filepath"
|
|
"runtime"
|
|
"strconv"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/spf13/cobra"
|
|
|
|
pad "github.com/PerpetualSoftware/pad"
|
|
"github.com/PerpetualSoftware/pad/internal/attachments"
|
|
"github.com/PerpetualSoftware/pad/internal/cli"
|
|
"github.com/PerpetualSoftware/pad/internal/cmdhelp"
|
|
"github.com/PerpetualSoftware/pad/internal/config"
|
|
|
|
"github.com/PerpetualSoftware/pad/internal/billing"
|
|
"github.com/PerpetualSoftware/pad/internal/collab"
|
|
"github.com/PerpetualSoftware/pad/internal/email"
|
|
"github.com/PerpetualSoftware/pad/internal/events"
|
|
"github.com/PerpetualSoftware/pad/internal/logging"
|
|
mcpserver "github.com/PerpetualSoftware/pad/internal/mcp"
|
|
"github.com/PerpetualSoftware/pad/internal/metrics"
|
|
"github.com/PerpetualSoftware/pad/internal/models"
|
|
oauthpkg "github.com/PerpetualSoftware/pad/internal/oauth"
|
|
"github.com/PerpetualSoftware/pad/internal/redisns"
|
|
"github.com/PerpetualSoftware/pad/internal/server"
|
|
"github.com/PerpetualSoftware/pad/internal/store"
|
|
"github.com/PerpetualSoftware/pad/internal/watchevents"
|
|
"github.com/PerpetualSoftware/pad/internal/webhooks"
|
|
"github.com/google/uuid"
|
|
mcptransport "github.com/mark3labs/mcp-go/server"
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
func serveCmd() *cobra.Command {
|
|
var host string
|
|
var port int
|
|
|
|
cmd := &cobra.Command{
|
|
Use: "start",
|
|
Short: "Start the Pad API server",
|
|
RunE: func(cmd *cobra.Command, args []string) error {
|
|
cfg := getConfig()
|
|
|
|
// Initialize structured logging
|
|
logLevel := os.Getenv("PAD_LOG_LEVEL")
|
|
if logLevel == "" {
|
|
logLevel = "info"
|
|
}
|
|
logFormat := os.Getenv("PAD_LOG_FORMAT")
|
|
if logFormat == "" {
|
|
logFormat = "text"
|
|
}
|
|
logging.Setup(logLevel, logFormat)
|
|
|
|
if cmd.Flags().Changed("host") {
|
|
cfg.Host = host
|
|
}
|
|
if cmd.Flags().Changed("port") {
|
|
cfg.Port = port
|
|
}
|
|
|
|
// Label pre-migration snapshots with this build's version, and
|
|
// honor the schema-ahead-guard escape hatch (--force or
|
|
// PAD_ALLOW_SCHEMA_AHEAD=1). See internal/store/migration_guard.go.
|
|
store.BinaryVersion = version
|
|
forceMigrate, _ := cmd.Flags().GetBool("force")
|
|
if forceMigrate || truthyEnv(os.Getenv("PAD_ALLOW_SCHEMA_AHEAD")) {
|
|
store.AllowSchemaAhead = true
|
|
}
|
|
|
|
// Open database (SQLite default, PostgreSQL via PAD_DB_DRIVER)
|
|
var s *store.Store
|
|
var err error
|
|
dbDriver := os.Getenv("PAD_DB_DRIVER")
|
|
if dbDriver == "postgres" {
|
|
pgURL := os.Getenv("PAD_DATABASE_URL")
|
|
if pgURL == "" {
|
|
return fmt.Errorf("PAD_DATABASE_URL is required when PAD_DB_DRIVER=postgres")
|
|
}
|
|
s, err = store.NewPostgres(pgURL)
|
|
if err != nil {
|
|
return fmt.Errorf("open postgres: %w", err)
|
|
}
|
|
slog.Info("Database using PostgreSQL")
|
|
} else {
|
|
s, err = store.New(cfg.DBPath)
|
|
if err != nil {
|
|
return fmt.Errorf("open database: %w", err)
|
|
}
|
|
slog.Info("Database using SQLite", "path", cfg.DBPath)
|
|
}
|
|
defer s.Close()
|
|
|
|
// Configure encryption key for sensitive fields (TOTP secrets).
|
|
// EnsureEncryptionKey resolves the key from (in order) env,
|
|
// config file, the persisted ~/.pad/encryption.key file, or a
|
|
// freshly-generated one written with 0600.
|
|
//
|
|
// Auto-generation is scoped to non-Postgres deployments.
|
|
// SQLite is single-instance by construction, so a generated
|
|
// local key is always correct. Postgres deployments may run
|
|
// multiple replicas behind a load balancer with separate
|
|
// filesystems — each replica would generate its own key and
|
|
// cross-replica decryption would fail. Operators MUST
|
|
// configure PAD_ENCRYPTION_KEY explicitly for Postgres; the
|
|
// repo's docker-compose.yml and deploy/k8s/secret.yaml both
|
|
// require it.
|
|
allowGenerate := dbDriver != "postgres"
|
|
if err := cfg.EnsureEncryptionKey(allowGenerate); err != nil {
|
|
return fmt.Errorf("encryption key: %w", err)
|
|
}
|
|
keyBytes, err := hex.DecodeString(cfg.EncryptionKey)
|
|
if err != nil || len(keyBytes) != 32 {
|
|
return fmt.Errorf("encryption key must be a 64-character hex string (32 bytes / 256 bits); got source=%q len=%d", cfg.EncryptionKeySource, len(cfg.EncryptionKey))
|
|
}
|
|
s.SetEncryptionKey(keyBytes)
|
|
switch cfg.EncryptionKeySource {
|
|
case "generated":
|
|
// Loud warning: operator should back this up and/or promote
|
|
// to a managed secret store in production. Logging at WARN
|
|
// level so it shows up in typical deployments that only
|
|
// surface warnings-and-above.
|
|
slog.Warn("Encryption key generated and persisted — back up the file, or set PAD_ENCRYPTION_KEY explicitly",
|
|
"path", cfg.EncryptionKeyFile())
|
|
case "file":
|
|
slog.Info("Encryption key loaded from file", "path", cfg.EncryptionKeyFile())
|
|
case "env":
|
|
slog.Info("Encryption key loaded from PAD_ENCRYPTION_KEY env var")
|
|
default:
|
|
slog.Info("Encryption key configured", "source", cfg.EncryptionKeySource)
|
|
}
|
|
|
|
// Backfill: encrypt any plaintext TOTP secrets.
|
|
if n, err := s.BackfillEncryptTOTPSecrets(); err != nil {
|
|
return fmt.Errorf("backfill TOTP encryption: %w", err)
|
|
} else if n > 0 {
|
|
slog.Info("Encrypted plaintext TOTP secrets", "count", n)
|
|
}
|
|
|
|
// Backfill: encrypt any plaintext webhook HMAC secrets (BUG-2057).
|
|
if n, err := s.EncryptWebhookSecretsAtRest(); err != nil {
|
|
return fmt.Errorf("encrypt webhook secrets at rest: %w", err)
|
|
} else if n > 0 {
|
|
slog.Info("Encrypted plaintext webhook secrets", "count", n)
|
|
}
|
|
|
|
// Backfill: populate item_wiki_links from existing item bodies
|
|
// (PLAN-1593 / TASK-1594). Idempotent — items already indexed
|
|
// at write time get a cheap EXISTS-skip; only newly-introduced
|
|
// items (e.g. from a fresh migration) actually parse. Failures
|
|
// here aren't fatal: a partial run leaves the table consistent
|
|
// and the next boot picks up where this one stopped.
|
|
if bf, err := s.BackfillWikiLinks(); err != nil {
|
|
slog.Warn("wiki-link backfill failed; non-fatal", "error", err)
|
|
} else if bf.ItemsIndexed > 0 {
|
|
slog.Info("Wiki-link backfill complete",
|
|
"items_scanned", bf.ItemsScanned,
|
|
"items_indexed", bf.ItemsIndexed,
|
|
"links_inserted", bf.LinksInserted,
|
|
"errors", bf.Errors,
|
|
)
|
|
} else if bf.ItemsScanned > 0 {
|
|
// Steady-state: scanned but nothing new to do.
|
|
slog.Debug("Wiki-link backfill no-op",
|
|
"items_scanned", bf.ItemsScanned, "errors", bf.Errors)
|
|
}
|
|
|
|
// Backfill: populate status_transitions from the historical
|
|
// activity log (PLAN-1628 / TASK-1637). Idempotent — gated on an
|
|
// empty table, so it replays history exactly once on the first
|
|
// boot after migration 063/042 and short-circuits thereafter (the
|
|
// write-path hook keeps the table populated). Non-fatal: a failed
|
|
// run leaves live data intact and only affects the pre-upgrade
|
|
// report window.
|
|
if st, err := s.BackfillStatusTransitions(); err != nil {
|
|
slog.Warn("status-transition backfill failed; non-fatal", "error", err)
|
|
} else if !st.Skipped && st.Inserted > 0 {
|
|
slog.Info("Status-transition backfill complete",
|
|
"activities_scanned", st.ActivitiesScanned,
|
|
"inserted", st.Inserted,
|
|
"errors", st.Errors,
|
|
)
|
|
}
|
|
|
|
// Auto-upgrade hook removed in IDEA-1479. The historical
|
|
// SeedDefaultCollections backfill was incompatible with templates
|
|
// that intentionally diverge from Defaults() (e.g. `blank`). Future
|
|
// changes that need to backfill collections into existing workspaces
|
|
// should be implemented as explicit migrations in
|
|
// internal/store/migrations/.
|
|
|
|
srv := server.New(s)
|
|
srv.SetVersion(version, commit, buildTime)
|
|
// PublicLinkBaseURL — not BaseURL() — so the server picks up
|
|
// PUBLIC_URL from the deployment env (BUG-899). BaseURL() is
|
|
// CLI-client-only and would leak the same env var into local
|
|
// CLI API routing on developer hosts.
|
|
srv.SetBaseURL(cfg.PublicLinkBaseURL())
|
|
srv.SetCORSOrigins(cfg.CORSOrigins)
|
|
srv.SetSecureCookies(cfg.SecureCookies)
|
|
srv.SetTrustedProxies(cfg.TrustedProxies)
|
|
srv.SetMetricsToken(cfg.MetricsToken)
|
|
srv.SetIPChangeEnforce(cfg.IPChangeEnforce)
|
|
srv.SetSSELimits(cfg.SSEMaxConnections, cfg.SSEMaxPerWorkspace, cfg.SSEMaxPerUser)
|
|
// Logged at startup so the BUG-2726 re-point is visible in an
|
|
// operator's own logs: PAD_SSE_MAX_CONNECTIONS now bounds
|
|
// /api/v1/events and /api/v1/events/stream TOGETHER, where it
|
|
// used to bound only the first. Someone who tuned it for one
|
|
// endpoint should see the effective numbers without having to
|
|
// read release notes to find out anything changed.
|
|
slog.Info("Stream connection limits (PER INSTANCE — not deployment-wide; there is no shared counter)",
|
|
"per_instance", cfg.SSEMaxConnections,
|
|
"per_workspace", cfg.SSEMaxPerWorkspace,
|
|
"per_principal", cfg.SSEMaxPerUser,
|
|
"per_instance_and_per_principal_cover", "/api/v1/events + /api/v1/events/stream",
|
|
"per_workspace_covers", "/api/v1/events")
|
|
|
|
// MCP tool-surface descriptor endpoint (PLAN-1888 / TASK-1891).
|
|
// Inject the cycle-free catalog→JSON serializer so the authed
|
|
// GET /api/v1/mcp/tool-surface route can serve it. Wired here
|
|
// (not in the cloud block) because the browser-side WebMCP layer
|
|
// needs the descriptors on BOTH cloud and self-host. Mirrors the
|
|
// SetMCPTransport injection: internal/server can't import
|
|
// internal/mcp (cycle), so cmd/pad — which imports both — hands
|
|
// the serializer down. Must be set before setupRouter runs.
|
|
srv.SetToolSurfaceHandler(mcpserver.ToolSurfaceJSON)
|
|
|
|
// Billing CTA gate (TASK-800). PAD_BILLING_AVAILABLE=true when
|
|
// the pad-cloud sidecar has Stripe keys configured so the web UI
|
|
// can show "Upgrade to Pro" buttons. Defaults to false so a fresh
|
|
// cloud deployment without Stripe doesn't expose dead-end CTAs.
|
|
if v := os.Getenv("PAD_BILLING_AVAILABLE"); v == "true" || v == "1" {
|
|
srv.SetBillingAvailable(true)
|
|
slog.Info("Billing CTAs enabled (PAD_BILLING_AVAILABLE)")
|
|
}
|
|
|
|
// Cloud-tenant mode: enable cloud-specific endpoints and
|
|
// behavior. Gated on IsCloudServer() (env-var opt-in) rather
|
|
// than IsCloud() (which is also true when a CLI user has
|
|
// picked "Cloud" as their `pad init` connection mode — that
|
|
// is a client signal, not a server-runtime signal).
|
|
if cfg.IsCloudServer() {
|
|
if cfg.CloudSecret == "" {
|
|
return fmt.Errorf("PAD_CLOUD_SECRET is required when running in cloud mode (PAD_MODE=cloud or PAD_CLOUD=true)")
|
|
}
|
|
// B7 (TASK-1932): fail fast rather than relying on the
|
|
// operator to always pair PAD_CLOUD with PAD_SECURE_COOKIES.
|
|
if err := cfg.ValidateCloudSecureCookies(); err != nil {
|
|
return err
|
|
}
|
|
srv.SetCloudMode(cfg.CloudSecret)
|
|
slog.Info("Cloud mode enabled")
|
|
|
|
// MCP Streamable HTTP transport (PLAN-943 TASK-950).
|
|
// Mount the public /mcp endpoint + RFC 9728 / RFC 8414
|
|
// discovery docs. Wired only in cloud mode because:
|
|
//
|
|
// - It requires user-owned PATs (the existing
|
|
// workspace-scoped PAT path is rejected — see
|
|
// MCPBearerAuth in middleware_mcp_auth.go), so a
|
|
// self-host running without users would have no
|
|
// usable auth path.
|
|
// - The discovery docs reference a public OAuth
|
|
// authorization server URL (TASK-951) that only
|
|
// exists on the Pad-Cloud deployment.
|
|
//
|
|
// Construction shape mirrors `pad mcp serve` (cmd/pad/mcp.go)
|
|
// minus the stdio runtime: cmdhelp.Doc → MCPServer →
|
|
// catalog/prompts/meta/resources registration → wrap in
|
|
// Streamable HTTP transport → hand to *server.Server.
|
|
// Resources are wired below via HTTPResourceFetcher
|
|
// (TASK-2101) — the in-process equivalent of the stdio
|
|
// ExecResourceFetcher that the original TASK-950 deferred
|
|
// because shelling out to the pad binary carries no
|
|
// per-OAuth-user credential context.
|
|
root := cmd.Root()
|
|
mcpDoc := cmdhelp.Build(root, root, cmdhelp.Options{
|
|
Binary: "pad",
|
|
Version: fullVersion(),
|
|
Homepage: padHomepage,
|
|
MaxDepth: -1,
|
|
})
|
|
mcpSrv := mcpserver.NewServer(mcpserver.Options{Version: fullVersion()})
|
|
// CurrentUserFromContext returns (*User, bool); the
|
|
// dispatcher's UserResolver signature is just
|
|
// (ctx) *User. The bool is "found", which equals
|
|
// "non-nil pointer" for the path the MCP middleware
|
|
// guarantees (it 401s on no-user before reaching
|
|
// dispatch), so flatten with a closure.
|
|
dispatcher := &mcpserver.HTTPHandlerDispatcher{
|
|
Handler: srv,
|
|
UserResolver: func(ctx context.Context) *models.User {
|
|
u, _ := server.CurrentUserFromContext(ctx)
|
|
return u
|
|
},
|
|
// OAuth-aware workspace lister (TASK-977).
|
|
// Filters error envelopes' available_workspaces
|
|
// hint by the token's consent allow-list so
|
|
// agents never see workspace slugs the user
|
|
// didn't explicitly grant.
|
|
Lister: mcpserver.NewOAuthWorkspaceLister(s),
|
|
// Tier-mismatch observability (TASK-1119).
|
|
// Bumps pad_mcp_authz_denials_total{reason="tier_mismatch"}
|
|
// when the dispatcher's per-tool scope check
|
|
// rejects a synthesized request. No-op until
|
|
// metrics are wired; safe to attach unconditionally
|
|
// because Server.RecordMCPTierMismatch nil-checks
|
|
// internally.
|
|
OnScopeDenied: srv.RecordMCPTierMismatch,
|
|
// PLAN-1933 DR-4: gate the remote MCP write path for
|
|
// unverified cloud users. /mcp mounts outside the
|
|
// /api/v1 stack, so the RequireVerifiedEmail HTTP
|
|
// middleware can't cover it — this hook is the
|
|
// perimeter's own gate (fires on mutating methods
|
|
// only, inside buildAuthedRequest). Cloud-only via
|
|
// srv.IsCloud(); a no-op on self-host.
|
|
RequireVerifiedEmail: func(user *models.User) bool {
|
|
return srv.IsCloud() && user != nil && !user.IsEmailVerified()
|
|
},
|
|
}
|
|
if _, regErr := mcpserver.Register(mcpSrv.MCP(), mcpserver.RegistryOptions{
|
|
Doc: mcpDoc,
|
|
// Shared multi-user state: this one stateless process
|
|
// dispatches for every OAuth user, so the session
|
|
// workspace must NEVER be trusted as a per-call
|
|
// resolution default — it would bleed across users /
|
|
// concurrent sessions (BUG-1865). NewSharedWorkspaceState
|
|
// makes ResolveDefault() always return "", forcing
|
|
// per-call explicit `workspace` (or the per-user
|
|
// maybeInjectWorkspace default). Local `pad mcp serve`
|
|
// (cmd/pad/mcp.go) keeps NewWorkspaceState — it's
|
|
// single-user-per-process and safe to inject.
|
|
Workspace: mcpserver.NewSharedWorkspaceState(),
|
|
Dispatcher: dispatcher,
|
|
PadVersion: fullVersion(),
|
|
}); regErr != nil {
|
|
return fmt.Errorf("register MCP catalog: %w", regErr)
|
|
}
|
|
mcpserver.RegisterPrompts(mcpSrv.MCP())
|
|
mcpserver.RegisterMeta(mcpSrv.MCP(), fullVersion())
|
|
|
|
// Read-only resource templates (TASK-2101). Parity with
|
|
// `pad mcp serve` (cmd/pad/mcp.go), which registers them via
|
|
// an ExecResourceFetcher. That fetcher shells out to the pad
|
|
// binary, inheriting one user's ~/.pad credentials — unusable
|
|
// in this shared multi-OAuth-user process. HTTPResourceFetcher
|
|
// is the in-process equivalent the original TASK-950 comment
|
|
// deferred: it dispatches each resource read through the same
|
|
// handler chain (reusing the dispatcher's user resolution +
|
|
// auth/consent perimeter), so the full resource set — including
|
|
// the bounded attachment image resource from PR #930 — is now
|
|
// available over remote /mcp.
|
|
mcpserver.RegisterResources(
|
|
mcpSrv.MCP(),
|
|
mcpserver.NewHTTPResourceFetcher(dispatcher),
|
|
nil, // no root flags on the remote transport (no --url)
|
|
)
|
|
// Stateless mode: every Streamable HTTP request stands
|
|
// alone, Bearer is the auth, no session resumption to
|
|
// manage. Matches the spike's verified shape.
|
|
// Stateless transport: every request stands alone; Bearer is the
|
|
// auth; no session resumption. mcp-go's WithStateLess(true)
|
|
// would do this exactly, BUT it wires StatelessSessionIdManager
|
|
// whose Generate() returns "" — so the response never carries
|
|
// Mcp-Session-Id, which makes the active-sessions tracker
|
|
// (TASK-1120) unobservable in production.
|
|
//
|
|
// Use a generate-only manager instead: every initialize gets a
|
|
// unique UUID on the response (so the tracker can key on it),
|
|
// but Validate accepts ANY incoming header value (including
|
|
// empty / arbitrary), so clients that never echo the
|
|
// session-id behave exactly as they did under the original
|
|
// WithStateLess(true) setup. Codex review on PR #400 round 1
|
|
// caught the gauge-stays-at-zero gap.
|
|
streamable := mcptransport.NewStreamableHTTPServer(
|
|
mcpSrv.MCP(),
|
|
mcptransport.WithEndpointPath("/mcp"),
|
|
mcptransport.WithSessionIdManager(&padMCPGenerateOnlySessionIDManager{}),
|
|
// mcp-go v0.56 turns on DNS-rebinding protection by
|
|
// default: a request whose accept socket is loopback but
|
|
// whose Host header is non-loopback gets a 403. pad-cloud's
|
|
// mcp.getpad.dev vhost sits behind a reverse proxy that
|
|
// forwards to this process over 127.0.0.1 while preserving
|
|
// the original Host, so the default would reject every real
|
|
// request. Disable it to keep the pre-v0.56 behaviour — the
|
|
// browser-driven rebinding threat it guards against doesn't
|
|
// apply here: this transport only mounts in cloud mode and
|
|
// every request is Bearer/OAuth-authenticated.
|
|
mcptransport.WithDisableLocalhostProtection(true),
|
|
)
|
|
// TASK-1120: optional env-driven overrides for the
|
|
// mcp-active-sessions tracker. Both default to the
|
|
// package values (30m TTL, 5m sweep) when unset. Must
|
|
// be applied BEFORE SetMCPTransport, which spawns the
|
|
// tracker — calling after has no effect.
|
|
srv.SetMCPSessionTrackerConfig(
|
|
parseDurationEnv("PAD_MCP_SESSION_TTL", 0),
|
|
parseDurationEnv("PAD_MCP_SESSION_SWEEP_INTERVAL", 0),
|
|
)
|
|
srv.SetMCPTransport(streamable, cfg.MCPPublicURL, cfg.AuthServerURL)
|
|
slog.Info("MCP /mcp transport mounted",
|
|
"public_url", cfg.MCPPublicURL,
|
|
"auth_server", cfg.AuthServerURL,
|
|
"resources_wired", true,
|
|
)
|
|
|
|
// OAuth 2.1 authorization server (PLAN-943 TASK-1024
|
|
// constructor + TASK-1025 HTTP handlers). Wired only
|
|
// when the cloud deployment has a configured MCP
|
|
// public URL — the OAuth server's audience strategy
|
|
// rejects every request unless tokens are bound to
|
|
// cfg.MCPPublicURL (the canonical resource URL the
|
|
// operator published).
|
|
//
|
|
// We treat cfg.MCPPublicURL as the canonical resource
|
|
// URL exactly — no path suffix appended. Per the MCP
|
|
// authorization spec the client compares the URL it
|
|
// was given against the discovery doc's `resource`
|
|
// field; a mismatch is a hard reject. So if the
|
|
// operator publishes "https://mcp.getpad.dev" (matches
|
|
// industry convention — mcp.stripe.com, mcp.linear.app,
|
|
// etc.), tokens are audience-bound to that exact
|
|
// string. The transport itself is internally mounted
|
|
// at /mcp on the chi router; pad-cloud's nginx router
|
|
// rewrites mcp.* root → /mcp transparently so external
|
|
// clients see a single canonical URL.
|
|
//
|
|
// HMAC secret reuses cfg.EncryptionKey, the same
|
|
// 32-byte hex key cloud deployments already require
|
|
// (validated above). fosite uses it to sign the
|
|
// opaque token signature half; rotation arrives
|
|
// alongside the operator runbook for TASK-953/954.
|
|
if cfg.MCPPublicURL != "" {
|
|
oauthSrv, oauthErr := oauthpkg.NewServer(oauthpkg.Config{
|
|
Store: s,
|
|
HMACSecret: keyBytes,
|
|
AllowedAudience: strings.TrimRight(cfg.MCPPublicURL, "/"),
|
|
})
|
|
if oauthErr != nil {
|
|
return fmt.Errorf("init OAuth server: %w", oauthErr)
|
|
}
|
|
srv.SetOAuthServer(oauthSrv)
|
|
// Same key powers stateless 6-digit claim codes
|
|
// (PLAN-1519 / TASK-1521 / IDEA-1517 §4). Reusing
|
|
// keyBytes here keeps the cloud-mode secret surface
|
|
// to a single 32-byte value the operator already
|
|
// rotates; the OAuth signing path and the claim
|
|
// HMAC path have the same rotation cadence + blast
|
|
// radius, so a shared secret is the right call.
|
|
srv.SetClaimSecret(keyBytes)
|
|
slog.Info("OAuth server mounted",
|
|
"endpoints", "/oauth/{register,authorize,token,claim}",
|
|
"audience", oauthSrv.AllowedAudience(),
|
|
)
|
|
|
|
// One-shot backfill of pre-TASK-1522 grant chains
|
|
// into oauth_connections + oauth_connection_workspaces
|
|
// (PLAN-1519 / TASK-1522 / IDEA-1517 §2). Idempotent:
|
|
// the inserts are ON CONFLICT DO NOTHING / INSERT OR
|
|
// IGNORE, so re-running on every startup is cheap and
|
|
// safe. The rewritten ListUserOAuthConnections reads
|
|
// from the new tables; running this BEFORE the HTTP
|
|
// server starts means /console/connected-apps never
|
|
// renders an empty page during the brief window
|
|
// between server-up and backfill-complete.
|
|
//
|
|
// Backfill failures don't abort startup — a partial
|
|
// run leaves the tables in a consistent state the
|
|
// next run completes (per-chain failures are logged
|
|
// and the run continues). The rewritten read path
|
|
// also has a defensive fallback for chains without
|
|
// connection rows, so any chain the backfill misses
|
|
// still renders.
|
|
if bf, bfErr := s.BackfillOAuthConnections(); bfErr != nil {
|
|
slog.Warn("oauth_connections backfill failed; non-fatal",
|
|
"error", bfErr)
|
|
} else if bf.ConnectionsCreated > 0 || bf.WorkspacesAdded > 0 {
|
|
slog.Info("oauth_connections backfill complete",
|
|
"chains_seen", bf.ChainsSeen,
|
|
"connections_created", bf.ConnectionsCreated,
|
|
"workspaces_added", bf.WorkspacesAdded,
|
|
"unresolved_slugs", bf.UnresolvedSlugs,
|
|
)
|
|
} else if bf.ChainsSeen > 0 {
|
|
// Quiet log on steady-state re-runs (chains
|
|
// scanned, nothing new to write) so ops can
|
|
// confirm the call ran without log noise.
|
|
slog.Debug("oauth_connections backfill no-op",
|
|
"chains_seen", bf.ChainsSeen)
|
|
}
|
|
} else {
|
|
slog.Warn("PAD_MCP_PUBLIC_URL not set — OAuth server NOT mounted (no canonical audience to bind tokens to)")
|
|
}
|
|
|
|
// Reverse pad → pad-cloud client (TASK-690). Used by
|
|
// handleDeleteAccount to cancel Stripe subscriptions + delete
|
|
// the Stripe customer before the local user row is purged.
|
|
// When PAD_CLOUD_SIDECAR_URL is unset we leave the sidecar
|
|
// hook nil — a cloud deploy without Stripe billing is
|
|
// unusual but valid (e.g. a staging instance), and in that
|
|
// case there's no upstream state to cancel.
|
|
//
|
|
// Outbound secret selection:
|
|
// pad-cloud validates inbound calls against a SINGLE secret
|
|
// (no rotation parsing on its side). During a rotation,
|
|
// operators roll pad first (so pad accepts both "new" and
|
|
// "old" inbound), then roll pad-cloud to the new key. Until
|
|
// pad-cloud has been rolled, it is still validating against
|
|
// the OLD secret — so the reverse call must send that old
|
|
// secret, not the new one.
|
|
//
|
|
// Resolution order:
|
|
// 1. PAD_CLOUD_OUTBOUND_SECRET explicitly set → use as-is.
|
|
// Correct for any rotation state when ops pin it.
|
|
// 2. Fall back to the LAST entry of PAD_CLOUD_SECRET (the
|
|
// older rotation value). This assumes the "new,old"
|
|
// convention during rollover and gracefully tracks the
|
|
// pad-cloud side without a separate env var. After
|
|
// rollover completes and CloudSecret collapses to a
|
|
// single value, first == last so it's still correct.
|
|
if cfg.CloudSidecarURL != "" {
|
|
outboundSecret := billing.ResolveOutboundSecret(cfg.CloudOutboundSecret, cfg.CloudSecret)
|
|
if outboundSecret == "" {
|
|
return fmt.Errorf("PAD_CLOUD_SIDECAR_URL is set but neither PAD_CLOUD_OUTBOUND_SECRET nor PAD_CLOUD_SECRET supplies a usable outbound secret")
|
|
}
|
|
srv.SetCloudSidecar(billing.NewCloudClient(cfg.CloudSidecarURL, outboundSecret))
|
|
outboundSource := "PAD_CLOUD_SECRET[last]"
|
|
if strings.TrimSpace(cfg.CloudOutboundSecret) != "" {
|
|
outboundSource = "PAD_CLOUD_OUTBOUND_SECRET"
|
|
}
|
|
slog.Info("Reverse pad-cloud sidecar wired", "url", cfg.CloudSidecarURL,
|
|
"outbound_source", outboundSource)
|
|
} else {
|
|
slog.Warn("PAD_CLOUD_SIDECAR_URL not set — account delete will NOT cancel Stripe subscriptions. Set this env var to cascade deletes.")
|
|
}
|
|
|
|
// Seed default plan limits (idempotent — won't overwrite admin changes)
|
|
if err := s.SeedPlanLimits(); err != nil {
|
|
return fmt.Errorf("seed plan limits: %w", err)
|
|
}
|
|
|
|
// Backfill: set existing users with empty plan to 'free'
|
|
// (first cloud-mode boot after upgrade from self-hosted)
|
|
if err := s.BackfillUserPlans("free"); err != nil {
|
|
slog.Warn("failed to backfill user plans", "error", err)
|
|
}
|
|
} else {
|
|
// Self-hosted mode: ensure all users have 'self-hosted' plan (no limits)
|
|
if err := s.BackfillUserPlans("self-hosted"); err != nil {
|
|
slog.Warn("failed to set self-hosted plans", "error", err)
|
|
}
|
|
}
|
|
|
|
// Wire attachment storage. Phase 1 = filesystem only; Phase 2 will
|
|
// register an "s3" backend alongside (or via a MigratingStore wrapping
|
|
// both during the cutover). Per-file cap is set by PAD_ATTACHMENT_MAX_BYTES
|
|
// or falls back to the package default (25 MiB).
|
|
attachDir := filepath.Join(cfg.DataDir, "attachments")
|
|
fsStore, err := attachments.NewFSStore(attachDir)
|
|
if err != nil {
|
|
return fmt.Errorf("init FS attachment store: %w", err)
|
|
}
|
|
attachReg := attachments.NewRegistry()
|
|
attachReg.Register(attachments.FSPrefix, fsStore)
|
|
var attachMax int64
|
|
if v := os.Getenv("PAD_ATTACHMENT_MAX_BYTES"); v != "" {
|
|
if n, perr := strconv.ParseInt(v, 10, 64); perr == nil && n > 0 {
|
|
attachMax = n
|
|
} else {
|
|
slog.Warn("PAD_ATTACHMENT_MAX_BYTES ignored — not a positive integer", "value", v)
|
|
}
|
|
}
|
|
srv.SetAttachments(attachReg, attachMax)
|
|
slog.Info("Attachment storage wired", "backend", "fs", "dir", attachDir)
|
|
|
|
// Workspace bundle import cap. Default is 2 GiB inside
|
|
// internal/server; PAD_IMPORT_BUNDLE_MAX_BYTES lets
|
|
// operators with larger exports raise the ceiling without
|
|
// recompiling (Codex review on PR #306 round 3).
|
|
if v := os.Getenv("PAD_IMPORT_BUNDLE_MAX_BYTES"); v != "" {
|
|
if n, perr := strconv.ParseInt(v, 10, 64); perr == nil && n > 0 {
|
|
srv.SetImportBundleMaxBytes(n)
|
|
slog.Info("Import bundle cap overridden", "max_bytes", n)
|
|
} else {
|
|
slog.Warn("PAD_IMPORT_BUNDLE_MAX_BYTES ignored — not a positive integer", "value", v)
|
|
}
|
|
}
|
|
|
|
// Single-artifact import cap. Default is 1 MiB inside
|
|
// internal/server; PAD_IMPORT_ARTIFACT_MAX_BYTES lets
|
|
// operators raise the ceiling without recompiling.
|
|
if v := os.Getenv("PAD_IMPORT_ARTIFACT_MAX_BYTES"); v != "" {
|
|
if n, perr := strconv.ParseInt(v, 10, 64); perr == nil && n > 0 {
|
|
srv.SetImportArtifactMaxBytes(n)
|
|
slog.Info("Import artifact cap overridden", "max_bytes", n)
|
|
} else {
|
|
slog.Warn("PAD_IMPORT_ARTIFACT_MAX_BYTES ignored — not a positive integer", "value", v)
|
|
}
|
|
}
|
|
|
|
// Orphan GC (TASK-886). Periodic sweep that reclaims
|
|
// attachments tombstoned past the grace period, plus
|
|
// uploads that were never associated with an item.
|
|
// Defaults: 24h interval, 30-day grace. Both override-
|
|
// able via env (e.g. PAD_ORPHAN_GC_INTERVAL=1m for tests
|
|
// where you want to see the sweep land within a CI run).
|
|
gcInterval := parseDurationEnv("PAD_ORPHAN_GC_INTERVAL", 0)
|
|
gcGrace := parseDurationEnv("PAD_ORPHAN_GC_GRACE", 0)
|
|
if gcInterval != 0 || gcGrace != 0 {
|
|
srv.SetOrphanGCConfig(gcInterval, gcGrace)
|
|
}
|
|
srv.StartOrphanGC()
|
|
|
|
// Wire the image processor used for thumbnail derivation
|
|
// (TASK-878) and the editor's rotate/crop tools (TASK-879/880).
|
|
// The default build picks the pure-Go backend (no cgo);
|
|
// `-tags libvips` will swap in the native backend in Phase 2.
|
|
//
|
|
// NewProcessor returns nil on the libvips build until Phase 2
|
|
// lands the real implementation — the server runs degraded
|
|
// (no thumbnail derivation, capabilities endpoint reports
|
|
// empty formats), but the binary boots cleanly. Skipping
|
|
// SetImageProcessor when the processor is nil keeps the
|
|
// wired-vs-unwired states cleanly distinct.
|
|
if imgProc := attachments.NewProcessor(); imgProc != nil {
|
|
srv.SetImageProcessor(imgProc)
|
|
slog.Info("Image processor wired", "formats", imgProc.Capabilities().ImageFormats)
|
|
} else {
|
|
slog.Info("Image processor not wired — thumbnail derivation disabled for this build")
|
|
}
|
|
|
|
// Initialize Prometheus metrics
|
|
m := metrics.New()
|
|
m.RegisterDBCollector(s.DB())
|
|
srv.SetMetrics(m)
|
|
slog.Info("Prometheus metrics enabled at /metrics")
|
|
|
|
// Attach event bus for real-time SSE. The client built here is
|
|
// shared with the watch bus below (BUG-2651) — nil when this is
|
|
// a single-instance deployment.
|
|
// ONE namespace value for all three Redis keyspaces (BUG-2724).
|
|
// Built here, before any of them, and passed into each
|
|
// constructor. THIS IS THE CONVENTION, NOT A GUARANTEE — each
|
|
// constructor takes its own Keys and nothing in the type
|
|
// system stops a future edit passing a different one;
|
|
// redis_keyspace_wiring_test.go enforces it by reading this
|
|
// file, which is a weaker instrument than a compiler.
|
|
//
|
|
// Validated eagerly: a namespace that cannot be used is a
|
|
// startup error, not a set of oddly-named keys discovered
|
|
// later.
|
|
redisKeys, err := redisns.Parse(cfg.RedisNamespace)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid PAD_REDIS_NAMESPACE: %w", err)
|
|
}
|
|
|
|
var eventBus events.EventBus
|
|
var watchRedis *redis.Client
|
|
if redisURL := os.Getenv("PAD_REDIS_URL"); redisURL != "" {
|
|
opts, err := redis.ParseURL(redisURL)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid PAD_REDIS_URL: %w", err)
|
|
}
|
|
rc := redis.NewClient(opts)
|
|
if err := rc.Ping(context.Background()).Err(); err != nil {
|
|
return fmt.Errorf("redis connection failed: %w", err)
|
|
}
|
|
// Route go-redis's own diagnostics into slog (BUG-2727).
|
|
// Its default logger writes to the standard log package,
|
|
// so its messages bypass the structured pipeline entirely
|
|
// — including the one that matters most here: "channel is
|
|
// full ... message is dropped", emitted when a
|
|
// subscription's 100-deep buffer stays full past the 60s
|
|
// send timeout. That is a silent notification loss whose
|
|
// only trace was a line nobody's log aggregator was
|
|
// shaped to catch.
|
|
redis.SetLogger(redisSlogLogger{})
|
|
eventBus = newObservedEventBus(cfg, rc, redisKeys, m)
|
|
watchRedis = rc
|
|
// THE EFFECTIVE PHASE IS LOGGED, not just configured
|
|
// (BUG-2736, codex round 9). An operator reading
|
|
// pad_event_sequence_resets_total cannot interpret it without
|
|
// knowing which phase this instance publishes in — a
|
|
// counter_backward rate is expected on phase 1 and an anomaly
|
|
// on phase 2 — and the config can arrive from an env var, a
|
|
// TOML file, or neither.
|
|
phase := 1
|
|
if cfg.EventsPublishEpoch {
|
|
phase = 2
|
|
}
|
|
// The heartbeat rollout is logged for the SAME reason and is a
|
|
// SEPARATE, independently-flipped migration (BUG-2738): an
|
|
// instance can be on id-space phase 2 and heartbeat phase 1 or
|
|
// any other combination. pad_event_subscription_cycled_total
|
|
// reads differently per phase — on heartbeat phase 1 a QUIET
|
|
// workspace's wedged route is undetectable, so a zero there
|
|
// means less than it does on phase 2.
|
|
heartbeatPhase := 1
|
|
if cfg.EventsHeartbeat {
|
|
heartbeatPhase = 2
|
|
}
|
|
slog.Info("Event bus using Redis pub/sub", "addr", opts.Addr, "db", opts.DB,
|
|
"namespace", redisKeys.Namespace(), "id_space_phase", phase,
|
|
"heartbeat_phase", heartbeatPhase)
|
|
} else {
|
|
eventBus = newObservedEventBus(cfg, nil, redisKeys, m)
|
|
slog.Info("Event bus using in-memory (single instance)")
|
|
}
|
|
// Wrap event bus with Prometheus instrumentation
|
|
eventBus = metrics.NewInstrumentedBus(eventBus, m)
|
|
srv.SetEventBus(eventBus)
|
|
|
|
// Watch/nudge notification bus (TASK-2533), on the SAME
|
|
// PAD_REDIS_URL switch as the event bus above (BUG-2651).
|
|
// Reusing that client rather than dialing a second one: they
|
|
// address the same server, and one connection pool with two
|
|
// logical channels is the shape events already assumes.
|
|
//
|
|
// The branch is here rather than inside the package because a
|
|
// self-hosted single-process binary should keep MemoryBus and
|
|
// never touch Redis at all — see internal/watchevents' package
|
|
// doc for what each implementation does and does not fix.
|
|
var watchBus watchevents.Bus
|
|
if watchRedis != nil {
|
|
redisWatchBus := watchevents.NewRedisBusWithKeys(watchRedis, watchevents.DefaultReplayBufferSize, redisKeys)
|
|
// Operational instrumentation (BUG-2727). Attached to the
|
|
// concrete type because the conditions it reports —
|
|
// dropped notifications, sequence gaps, id-space resets,
|
|
// the receive loop stopping — are detected inside the bus
|
|
// and are not visible at the Bus interface, so a wrapper
|
|
// of the events.EventBus kind cannot see them.
|
|
redisWatchBus.SetObserver(metrics.NewWatchEventsObserver(m))
|
|
watchBus = redisWatchBus
|
|
slog.Info("Watch notification bus using Redis pub/sub")
|
|
} else {
|
|
memWatchBus := watchevents.New()
|
|
memWatchBus.SetObserver(metrics.NewWatchEventsObserver(m))
|
|
watchBus = memWatchBus
|
|
slog.Info("Watch notification bus using in-memory (single instance)")
|
|
}
|
|
srv.SetWatchEventsBus(watchBus)
|
|
|
|
// Live-session presence registry (PLAN-2558 S1), on the SAME
|
|
// PAD_REDIS_URL switch as both buses above (BUG-2698). The
|
|
// three now cross instance boundaries together, which is the
|
|
// point: a shared bus with a per-process registry was worse
|
|
// than either being consistent, because a session-targeted
|
|
// push was resolved against the answering instance's view and
|
|
// skipped the publish for a session the bus could have
|
|
// reached.
|
|
//
|
|
// A self-hosted single-process binary keeps the in-memory
|
|
// registry and never touches Redis, same as the buses.
|
|
var sessionPresence server.SessionPresence
|
|
// Declared out here so the shutdown sequence below can close it
|
|
// at the right point in the order — see there for why that point
|
|
// is before http.Server.Shutdown rather than after.
|
|
var redisPresence *server.RedisSessionPresence
|
|
if watchRedis != nil {
|
|
redisPresence = server.NewRedisSessionPresenceWithKeys(watchRedis, redisKeys)
|
|
// Presence is fail-soft by design — a failed write risks
|
|
// leaving a live session unlisted rather than dropping
|
|
// its connection — so its failures have no user-visible
|
|
// signal beyond a push that quietly reaches fewer
|
|
// sessions. The counter is the only alertable trace
|
|
// (BUG-2727).
|
|
redisPresence.SetFailureObserver(func(op string) {
|
|
m.SessionPresenceFailuresTotal.WithLabelValues(op).Inc()
|
|
})
|
|
// Backstop for early-return paths that never reach the
|
|
// shutdown sequence; Close is idempotent. Deliberately does
|
|
// NOT delete this instance's entries — a shutdown racing a
|
|
// reconnect elsewhere would then delete a session that had
|
|
// already re-registered — so they clear on their TTL.
|
|
defer redisPresence.Close()
|
|
sessionPresence = redisPresence
|
|
slog.Info("Session presence registry using Redis (shared across instances)")
|
|
} else {
|
|
sessionPresence = server.NewMemorySessionPresence()
|
|
slog.Info("Session presence registry using in-memory (single instance)")
|
|
}
|
|
srv.SetSessionPresence(sessionPresence)
|
|
|
|
// Redis reachability prober (BUG-2727). Reports into
|
|
// /health/ready's payload and pad_redis_up; deliberately does
|
|
// NOT gate readiness — see server.RedisHealth's doc comment.
|
|
// Only wired when there IS a Redis, so a single-process
|
|
// deployment reports no redis block rather than a false one.
|
|
if watchRedis != nil {
|
|
// Registered here rather than in metrics.New so a
|
|
// Redis-less deployment exports no pad_redis_up series at
|
|
// all — a permanent 0 would read as an outage.
|
|
m.RegisterRedisUp()
|
|
redisHealth := server.NewRedisHealth(watchRedis, func(ok bool) {
|
|
if ok {
|
|
m.RedisUp.Set(1)
|
|
} else {
|
|
m.RedisUp.Set(0)
|
|
}
|
|
})
|
|
redisHealth.Start()
|
|
defer redisHealth.Stop()
|
|
srv.SetRedisHealth(redisHealth)
|
|
}
|
|
|
|
// Yjs collab room manager (PLAN-1248). Single-instance only
|
|
// today; multi-replica fanout via Redis is a deferred IDEA.
|
|
// MemoryOpBus is in-process; the OpBus interface keeps the
|
|
// door open for a RedisOpBus drop-in later.
|
|
collabBus := collab.NewMemoryOpBus()
|
|
srv.SetCollabRoomManager(collab.NewRoomManager(s, collabBus))
|
|
slog.Info("Collab room manager wired (Yjs over /api/v1/collab/{itemID})")
|
|
|
|
// Op-log prune sweeper (TASK-1309). Periodic background
|
|
// loop that deletes Yjs op-log rows older than minAge.
|
|
// Defaults: 1h interval, 24h minAge — both override-able
|
|
// via env (PAD_OPLOG_GC_INTERVAL=5m for tests where you
|
|
// want to see the sweep land within a CI run, etc.).
|
|
oplogGCInterval := parseDurationEnv("PAD_OPLOG_GC_INTERVAL", 0)
|
|
oplogGCMinAge := parseDurationEnv("PAD_OPLOG_GC_MIN_AGE", 0)
|
|
if oplogGCInterval != 0 || oplogGCMinAge != 0 {
|
|
srv.SetOpLogGCConfig(oplogGCInterval, oplogGCMinAge)
|
|
}
|
|
srv.StartOpLogGC()
|
|
|
|
// Token reaper (PLAN-1933 DR-5 / TASK-1936). Periodic sweep
|
|
// that deletes expired/used email-verification tokens,
|
|
// password-reset tokens, sessions, and CLI-auth sessions —
|
|
// the CleanExpired* methods existed but were never called.
|
|
// Default: 1h interval, override-able via env
|
|
// (PAD_TOKEN_REAPER_INTERVAL=1m for tests/CI).
|
|
if reaperInterval := parseDurationEnv("PAD_TOKEN_REAPER_INTERVAL", 0); reaperInterval != 0 {
|
|
srv.SetTokenReaperConfig(reaperInterval)
|
|
}
|
|
srv.StartTokenReaper()
|
|
|
|
// Workspace hard-purge sweeper (TASK-1966). Periodic sweep
|
|
// that hard-deletes workspaces soft-deleted more than 30 days
|
|
// ago — cascading every child row and reclaiming attachment
|
|
// blobs — to honor the /privacy 30-day GDPR erasure SLA.
|
|
// DeleteAccountAtomic / DeleteWorkspace only soft-delete
|
|
// (workspaces.deleted_at); nothing else ever expunges them.
|
|
// Defaults: 24h interval, 30-day retention — both override-
|
|
// able via env (PAD_WORKSPACE_PURGE_INTERVAL=1m /
|
|
// PAD_WORKSPACE_PURGE_RETENTION=1s for tests/CI). Must run
|
|
// AFTER the attachment registry is wired (above) so blob
|
|
// reclamation has a backend.
|
|
wsPurgeInterval := parseDurationEnv("PAD_WORKSPACE_PURGE_INTERVAL", 0)
|
|
wsPurgeRetention := parseDurationEnv("PAD_WORKSPACE_PURGE_RETENTION", 0)
|
|
if wsPurgeInterval != 0 || wsPurgeRetention != 0 {
|
|
srv.SetWorkspacePurgeConfig(wsPurgeInterval, wsPurgeRetention)
|
|
}
|
|
srv.StartWorkspacePurgeSweeper()
|
|
|
|
// Attach webhook dispatcher for outgoing notifications
|
|
srv.SetWebhookDispatcher(webhooks.NewDispatcher(s))
|
|
|
|
// SPEC-3 event outbox drain (TASK-2714). Started AFTER the
|
|
// dispatcher is attached: a drain running without one acks its
|
|
// rows as having nowhere to go, which for the first few seconds
|
|
// of a boot would silently discard real events.
|
|
outboxInterval := parseDurationEnv("PAD_OUTBOX_DRAIN_INTERVAL", 0)
|
|
outboxLease := parseDurationEnv("PAD_OUTBOX_CLAIM_LEASE", 0)
|
|
outboxRetention := parseDurationEnv("PAD_OUTBOX_RETENTION", 0)
|
|
outboxMaxAge := parseDurationEnv("PAD_OUTBOX_MAX_AGE", 0)
|
|
if outboxInterval != 0 || outboxLease != 0 || outboxRetention != 0 || outboxMaxAge != 0 {
|
|
srv.SetOutboxDrainConfig(outboxInterval, outboxLease, outboxRetention, outboxMaxAge, 0)
|
|
}
|
|
srv.StartOutboxDrain()
|
|
|
|
// Attach email sender: env vars first, then platform settings overlay
|
|
if cfg.MailerooAPIKey != "" {
|
|
fromAddr := cfg.EmailFrom
|
|
if fromAddr == "" {
|
|
fromAddr = "noreply@getpad.dev"
|
|
}
|
|
fromName := cfg.EmailFromName
|
|
if fromName == "" {
|
|
fromName = "Pad"
|
|
}
|
|
// PublicLinkBaseURL — emailed links must use the deployment's
|
|
// public URL (PUBLIC_URL), not the CLI BaseURL().
|
|
srv.SetEmailSender(email.NewSender(cfg.MailerooAPIKey, fromAddr, fromName, cfg.PublicLinkBaseURL()), cfg.MailerooAPIKey)
|
|
slog.Info("Email sending enabled via Maileroo (env)")
|
|
}
|
|
// Platform settings can override or provide email config
|
|
srv.InitEmailFromSettings()
|
|
|
|
// Initialize 2FA challenge signing key (persisted in platform_settings)
|
|
if err := srv.Init2FASecret(); err != nil {
|
|
return fmt.Errorf("init 2FA secret: %w", err)
|
|
}
|
|
|
|
// Mount embedded web UI if available
|
|
webFS, err := fs.Sub(pad.WebUI, "web/build")
|
|
if err == nil {
|
|
if entries, err := fs.ReadDir(webFS, "."); err == nil && len(entries) > 0 {
|
|
srv.SetWebUI(webFS)
|
|
slog.Info("Serving embedded web UI")
|
|
}
|
|
}
|
|
|
|
// First-run bootstrap token (TASK-1167 / PLAN-1166) +
|
|
// PAD_BYPASS_SETUP_TOKEN open-mode escape hatch.
|
|
//
|
|
// The bypass env-var lets operators on trusted networks
|
|
// (Unraid behind a firewall, Tailscale-only deployments)
|
|
// claim the first admin via the web UI without copying a
|
|
// token out of `docker logs`. Cloud mode always ignores it.
|
|
//
|
|
// Branches by current state:
|
|
//
|
|
// 1. UserCount > 0 → mop up any stale token file left by a
|
|
// previous successful bootstrap whose os.Remove somehow
|
|
// failed (D4). No banner.
|
|
// 2. UserCount == 0 + self-host + bypass on → wire the
|
|
// bypass into the server, skip token generation
|
|
// entirely (no .bootstrap-token file written), log a
|
|
// distinct open-mode banner so the operator sees
|
|
// explicitly that the surface is unprotected.
|
|
// 3. UserCount == 0 + self-host + bypass off → ensure a
|
|
// token exists, load it into the server, log the
|
|
// banner so the operator can grab it from `docker
|
|
// logs`. Failures are WARN-and-continue (D7) — never
|
|
// abort startup.
|
|
// 4. UserCount == 0 + cloud mode → no-op (D10). Cloud
|
|
// bootstrap stays loopback-only regardless of bypass.
|
|
bypassEnv := os.Getenv("PAD_BYPASS_SETUP_TOKEN")
|
|
bypassSetupToken := bypassEnv == "true" || bypassEnv == "1"
|
|
srv.SetBypassSetupToken(bypassSetupToken && !cfg.IsCloudServer())
|
|
if userCount, ucErr := s.UserCount(); ucErr != nil {
|
|
slog.Warn("could not check user count for bootstrap token wiring", "error", ucErr)
|
|
} else if userCount > 0 {
|
|
if cleanupErr := server.CleanupStaleBootstrapToken(cfg.DataDir); cleanupErr != nil {
|
|
slog.Warn("stale bootstrap token cleanup failed", "error", cleanupErr)
|
|
}
|
|
} else if !cfg.IsCloudServer() {
|
|
if bypassSetupToken {
|
|
// Don't generate or persist a token; the surface is
|
|
// already open and a token file would just be a
|
|
// confusing artifact for the operator.
|
|
if cleanupErr := server.CleanupStaleBootstrapToken(cfg.DataDir); cleanupErr != nil {
|
|
slog.Warn("stale bootstrap token cleanup failed", "error", cleanupErr)
|
|
}
|
|
logOpenBootstrapBanner(cfg)
|
|
} else {
|
|
token, tokenPath, terr := server.EnsureBootstrapToken(cfg.DataDir)
|
|
if terr != nil {
|
|
slog.Warn("first-run bootstrap token unavailable; falling back to loopback-only setup",
|
|
"error", terr,
|
|
"hint", "operator can run `pad auth setup` from inside the container")
|
|
} else {
|
|
srv.SetBootstrapToken(token, tokenPath)
|
|
logBootstrapBanner(token, cfg, tokenPath)
|
|
}
|
|
}
|
|
} else if bypassSetupToken {
|
|
// Cloud mode + bypass: defensively log a warning so a
|
|
// misconfigured operator notices their flag was ignored.
|
|
slog.Warn("PAD_BYPASS_SETUP_TOKEN is ignored in cloud mode (PAD_CLOUD/PAD_MODE=cloud)",
|
|
"hint", "cloud bootstrap stays loopback-only by design")
|
|
}
|
|
|
|
// Graceful shutdown: listen for SIGINT/SIGTERM
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
// Start server in a goroutine
|
|
errCh := make(chan error, 1)
|
|
go func() {
|
|
errCh <- srv.ListenAndServe(cfg.Addr())
|
|
}()
|
|
|
|
// Wait for signal or server error
|
|
select {
|
|
case err := <-errCh:
|
|
// Server failed to start or crashed
|
|
return err
|
|
case <-ctx.Done():
|
|
// Received shutdown signal
|
|
slog.Info("Shutting down server (30s grace period)...")
|
|
stop() // Reset signal handling so a second signal force-kills
|
|
|
|
// Close event bus first — this terminates SSE handler
|
|
// goroutines so http.Server.Shutdown won't block on them.
|
|
// eventBus is always non-nil here: assigned a few lines
|
|
// above to a concrete *metrics.InstrumentedBus return value.
|
|
eventBus.Close()
|
|
slog.Info("Event bus closed")
|
|
|
|
// Presence closes FIRST — before the watch bus, and well before
|
|
// Shutdown. The ordering is load-bearing twice over (codex
|
|
// rounds 4 and 5 on BUG-2698).
|
|
//
|
|
// Remove waits for a session's renewal goroutine. Closing the bus
|
|
// is what RELEASES the SSE handlers, so each one immediately runs
|
|
// its deferred Remove — and any Remove that runs before Close
|
|
// still finds a live renewal to wait for, bypassing the drain
|
|
// bound entirely and putting that wait in front of Shutdown.
|
|
// Closing presence first cancels every renewal and drains them,
|
|
// so the Removes that follow have nothing left to wait on.
|
|
//
|
|
// Note it does NOT empty the registry — an earlier version of
|
|
// this comment said so, and Close deliberately retains the
|
|
// entries precisely so a concurrent Remove can still find and
|
|
// await a renewal (codex rounds 6 and 10). Remove's own wait is
|
|
// bounded by the same drain deadline, so a renewal Close could
|
|
// not drain cannot hold Shutdown either.
|
|
//
|
|
// Nothing here depends on the bus, so there is no cost to going
|
|
// first. Idempotent, so the deferred Close at the wiring site
|
|
// stays a harmless backstop for early-return paths.
|
|
if redisPresence != nil {
|
|
redisPresence.Close()
|
|
slog.Info("Session presence registry closed")
|
|
}
|
|
|
|
// Same reasoning for the watch bus, and it matters for the
|
|
// same reason: GET /api/v1/events/stream is a long-lived
|
|
// handler blocked on this bus's channel, so leaving it open
|
|
// keeps http.Server.Shutdown waiting out its full 30s
|
|
// deadline (codex round 2 on BUG-2651). Closing here rather
|
|
// than only in srv.Stop() below also tears the Redis
|
|
// subscription down promptly instead of at the very end.
|
|
// Both Close implementations are idempotent, so the second
|
|
// call inside srv.Stop() is a no-op.
|
|
//
|
|
// THE TRADE, which is the same one eventBus above already
|
|
// makes and is worth naming: closing BEFORE Shutdown drains
|
|
// handlers means a push already in flight publishes into a
|
|
// closed bus and is not delivered. Closing AFTER instead would
|
|
// hold Shutdown for its full 30s deadline on any open stream,
|
|
// every time. Neither is free; this side loses a message in a
|
|
// window measured in milliseconds during a deliberate shutdown,
|
|
// the other side delays every shutdown by half a minute.
|
|
//
|
|
// WHAT IT NO LONGER COSTS is the caller's understanding of it
|
|
// (BUG-2699). This comment used to end "...and still return HTTP
|
|
// 200 with pushed:true", and then note that fixing it needed an
|
|
// interface change and a different unit. That unit landed:
|
|
// Bus.Publish reports acceptance, so a push publishing into a
|
|
// closed bus gets ErrBusClosed and answers 503. The message is
|
|
// still lost; the caller is no longer told it was sent, and can
|
|
// safely re-send once the server is back.
|
|
watchBus.Close()
|
|
slog.Info("Watch notification bus closed")
|
|
|
|
shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
if err := srv.Shutdown(shutdownCtx); err != nil {
|
|
slog.Error("HTTP server shutdown error", "error", err)
|
|
}
|
|
|
|
// http.Server.Shutdown doesn't terminate hijacked
|
|
// connections (WebSockets), so collab sessions keep
|
|
// running until something explicitly closes them.
|
|
// srv.Stop() runs the collab RoomManager.Close path
|
|
// (TASK-1255) plus the other long-running background
|
|
// loops (orphan GC, MCP audit writer, etc.). Without
|
|
// this, an open collab WS would race the deferred
|
|
// store close on process exit.
|
|
srv.Stop()
|
|
|
|
slog.Info("Server stopped")
|
|
return nil
|
|
}
|
|
},
|
|
}
|
|
|
|
cmd.Flags().StringVar(&host, "host", "127.0.0.1", "host address to listen on")
|
|
cmd.Flags().IntVar(&port, "port", 7777, "port to listen on")
|
|
cmd.Flags().Bool("force", false, "start even if the database schema is newer than this binary (a downgrade — risks data corruption; see the README \"Upgrading Pad\" section)")
|
|
|
|
return cmd
|
|
}
|
|
|
|
// logBootstrapBanner emits a single multi-line INFO log entry pointing the
|
|
// operator at the /setup#token=<x> URL they need to visit to claim the
|
|
// first admin. The banner is greppable by "Pad first-run setup" so log
|
|
// aggregators can surface it. Token is in a URL fragment (#token=) so it
|
|
// is never transmitted to the server in HTTP requests — only the
|
|
// browser sees it. The frontend strips it from the URL via
|
|
// history.replaceState on mount and submits via the X-Bootstrap-Token
|
|
// header.
|
|
//
|
|
// See TASK-1167 / PLAN-1166. Called only on first start with zero
|
|
// users in self-host mode.
|
|
func logBootstrapBanner(token string, cfg *config.Config, tokenPath string) {
|
|
host := cfg.Host
|
|
switch host {
|
|
case "", "0.0.0.0", "::", "[::]":
|
|
// PAD_HOST not bound to a specific interface — render as
|
|
// <your-host> placeholder in the banner since we don't know
|
|
// which network interface the operator wants to reach this
|
|
// instance from.
|
|
host = "<your-host>"
|
|
}
|
|
url := fmt.Sprintf("http://%s:%d/setup#token=%s", host, cfg.Port, token)
|
|
banner := fmt.Sprintf(`
|
|
========================================================================
|
|
Pad first-run setup
|
|
========================================================================
|
|
|
|
No users exist yet. To create the first admin account, visit:
|
|
|
|
%s
|
|
|
|
This token is one-time. After the first admin is created the token
|
|
is consumed and this banner stops appearing.
|
|
|
|
To regenerate, delete %s and restart.
|
|
|
|
========================================================================
|
|
`, url, tokenPath)
|
|
|
|
// BUG-1182: bypass slog for the banner. slog's text handler is
|
|
// contractually one-line-per-record and escapes literal newlines as
|
|
// `\n`, which renders the multi-line banner as a single wide line in
|
|
// `docker logs` — exactly the surface where operators look for it.
|
|
// Banner-style operator output isn't structured logging; stderr is the
|
|
// conventional channel for it (kubectl / docker / helm / systemd all
|
|
// do this). docker logs captures stderr alongside stdout, so the
|
|
// banner stays visible.
|
|
fmt.Fprint(os.Stderr, banner)
|
|
|
|
// Companion structured log so log aggregators that parse slog JSON
|
|
// still record the event. Deliberately does NOT include the URL or
|
|
// token — those are in the stderr banner where the operator looks
|
|
// for them. Repeating the URL as a parseable structured field would
|
|
// give log aggregators an easy-to-extract token, which is precisely
|
|
// what the URL-fragment design (TASK-1167 F10) is trying to avoid.
|
|
// Operators / agents that want the token programmatically should
|
|
// read the token_path file directly.
|
|
slog.Info("first-run bootstrap setup banner emitted to stderr — see container logs",
|
|
"token_path", tokenPath)
|
|
}
|
|
|
|
// logOpenBootstrapBanner is the PAD_BYPASS_SETUP_TOKEN companion to
|
|
// logBootstrapBanner. When the operator opts into the bypass, no token
|
|
// is generated — the bootstrap endpoint accepts the first-admin POST
|
|
// from any IP without an X-Bootstrap-Token header. The banner makes
|
|
// the security trade-off explicit so an operator who set the flag
|
|
// without thinking through the implications sees an obvious WARN.
|
|
//
|
|
// Self-host only; cmd/pad/main.go gates this call behind the
|
|
// !cfg.IsCloudServer() branch.
|
|
func logOpenBootstrapBanner(cfg *config.Config) {
|
|
host := cfg.Host
|
|
switch host {
|
|
case "", "0.0.0.0", "::", "[::]":
|
|
host = "<your-host>"
|
|
}
|
|
url := fmt.Sprintf("http://%s:%d/setup", host, cfg.Port)
|
|
banner := fmt.Sprintf(`
|
|
========================================================================
|
|
Pad first-run setup (open mode)
|
|
========================================================================
|
|
|
|
PAD_BYPASS_SETUP_TOKEN is set. The first admin can be created
|
|
directly from the web UI — no bootstrap token required:
|
|
|
|
%s
|
|
|
|
WARNING: anyone who can reach this URL can claim the first admin
|
|
account until you create one. Only leave this enabled on networks
|
|
you trust (LAN behind a firewall, Tailscale, etc.).
|
|
|
|
To re-enable token-protected setup, unset PAD_BYPASS_SETUP_TOKEN
|
|
and restart. Once a user exists this banner stops appearing.
|
|
|
|
========================================================================
|
|
`, url)
|
|
fmt.Fprint(os.Stderr, banner)
|
|
slog.Warn("first-run bootstrap is OPEN (PAD_BYPASS_SETUP_TOKEN=true) — anyone reachable on the WebUI port can claim the first admin until one is created")
|
|
}
|
|
|
|
// --- stop ---
|
|
|
|
func stopCmd() *cobra.Command {
|
|
return &cobra.Command{
|
|
Use: "stop",
|
|
Short: "Stop the background Pad server",
|
|
RunE: func(cmd *cobra.Command, args []string) error {
|
|
cfg := getConfig()
|
|
if err := cli.StopServer(cfg); err != nil {
|
|
return err
|
|
}
|
|
fmt.Println("Server stopped.")
|
|
return nil
|
|
},
|
|
}
|
|
}
|
|
|
|
// --- open ---
|
|
|
|
func openCmd() *cobra.Command {
|
|
return &cobra.Command{
|
|
Use: "open",
|
|
Short: "Open the Pad web UI in your browser",
|
|
RunE: func(cmd *cobra.Command, args []string) error {
|
|
cfg := getConfiguredConfig()
|
|
if err := cli.EnsureServer(cfg); err != nil {
|
|
return fmt.Errorf("start server: %w", err)
|
|
}
|
|
|
|
url := cfg.BrowserURL()
|
|
|
|
// If there's a workspace, go directly to it
|
|
ws, _ := cli.DetectWorkspace(workspaceFlag)
|
|
if ws != "" {
|
|
url += "/" + ws
|
|
}
|
|
|
|
fmt.Printf("Opening %s\n", url)
|
|
return openBrowser(url)
|
|
},
|
|
}
|
|
}
|
|
|
|
func openBrowser(url string) error {
|
|
switch runtime.GOOS {
|
|
case "darwin":
|
|
return exec.Command("open", url).Start()
|
|
case "linux":
|
|
return exec.Command("xdg-open", url).Start()
|
|
case "windows":
|
|
return exec.Command("rundll32", "url.dll,FileProtocolHandler", url).Start()
|
|
default:
|
|
return fmt.Errorf("unsupported platform — open %s manually", url)
|
|
}
|
|
}
|
|
|
|
// --- auth commands ---
|
|
|
|
// padMCPGenerateOnlySessionIDManager is the SessionIdManager used for
|
|
// the /mcp Streamable HTTP transport (TASK-1120). It produces a fresh
|
|
// UUID on every initialize so the response carries Mcp-Session-Id
|
|
// (which the active-sessions tracker reads), but accepts ANY incoming
|
|
// session-id value — including empty strings — without rejection.
|
|
//
|
|
// This intentionally diverges from both shipped mcp-go managers:
|
|
//
|
|
// - StatelessSessionIdManager: Generate() returns "" → tracker can't
|
|
// observe sessions in production. Original choice; broken for
|
|
// observability after TASK-1120.
|
|
// - StatelessGeneratingSessionIdManager: Generate() works, BUT
|
|
// Validate() rejects clients that don't echo the spec'd UUID
|
|
// prefix → breaking change for any client previously running
|
|
// under WithStateLess(true) that didn't track session-id.
|
|
//
|
|
// We're truly stateless server-side (no DB, no map of session IDs to
|
|
// validate against), so accepting any incoming value is correct: the
|
|
// "session" is purely a per-request observability label and the
|
|
// server doesn't depend on any client honoring it. Termination is a
|
|
// no-op for the same reason — there's no per-session state to free
|
|
// when DELETE arrives, and the active-sessions tracker handles its
|
|
// own bookkeeping.
|
|
type padMCPGenerateOnlySessionIDManager struct{}
|
|
|
|
func (padMCPGenerateOnlySessionIDManager) Generate() string {
|
|
return "pad-mcp-" + uuid.NewString()
|
|
}
|
|
|
|
func (padMCPGenerateOnlySessionIDManager) Validate(string) (isTerminated bool, err error) {
|
|
return false, nil
|
|
}
|
|
|
|
func (padMCPGenerateOnlySessionIDManager) Terminate(string) (isNotAllowed bool, err error) {
|
|
return false, nil
|
|
}
|
|
|
|
// humanBytes formats a byte count with the smallest IEC unit that
|
|
// keeps the value under 1024 — matches the convention used across
|
|
// the web UI's storage bar so CLI and browser reads agree.
|
|
// parseDurationEnv reads a duration env var (Go syntax: 1h, 30m,
|
|
// 24h, 720h, etc). Returns the default when the var is unset; logs
|
|
// a warning and returns the default on a parse error so a typo
|
|
// doesn't silently break the GC schedule.
|
|
func parseDurationEnv(name string, def time.Duration) time.Duration {
|
|
v := os.Getenv(name)
|
|
if v == "" {
|
|
return def
|
|
}
|
|
d, err := time.ParseDuration(v)
|
|
if err != nil {
|
|
slog.Warn(name+" ignored — not a valid Go duration", "value", v, "error", err)
|
|
return def
|
|
}
|
|
return d
|
|
}
|
|
|
|
func humanBytes(n int64) string {
|
|
if n < 0 {
|
|
return "?"
|
|
}
|
|
const unit = 1024
|
|
if n < unit {
|
|
return fmt.Sprintf("%d B", n)
|
|
}
|
|
div, exp := int64(unit), 0
|
|
for x := n / unit; x >= unit; x /= unit {
|
|
div *= unit
|
|
exp++
|
|
}
|
|
suffixes := []string{"KiB", "MiB", "GiB", "TiB", "PiB"}
|
|
if exp >= len(suffixes) {
|
|
exp = len(suffixes) - 1
|
|
}
|
|
return fmt.Sprintf("%.1f %s", float64(n)/float64(div), suffixes[exp])
|
|
}
|
|
|
|
// newObservedEventBus builds the event bus for this deployment shape with its
|
|
// operational observer already attached (BUG-2731). A nil client selects the
|
|
// in-process bus.
|
|
//
|
|
// EXTRACTED SO THE WIRING IS TESTABLE, which is the whole point (CONVE-19:
|
|
// wiring is a claim). Inline in the command's RunE, the SetObserver call was a
|
|
// claim no test could reach: an events-package test proves the bus calls its
|
|
// observer and a metrics-package test proves the adapter maps it, and both
|
|
// pass with that line deleted.
|
|
//
|
|
// The observer attaches to the CONCRETE bus because SetObserver is not part of
|
|
// the EventBus interface the wrapper implements, so it has to happen before
|
|
// the value is widened — a typing constraint, not a subtle ordering
|
|
// requirement. What the wrapper genuinely cannot do is REPLACE this: the
|
|
// conditions reported here are detected inside the bus, on the receive path
|
|
// and inside the coverage rules, not at the interface it wraps.
|
|
//
|
|
// The in-process bus is wired too, deliberately: a single-process deployment
|
|
// restarts, and the cold-buffer resume gap is exactly as real there. Its reset
|
|
// counter stays at zero by construction — MemoryBus owns its own IDs and has
|
|
// no shared counter to lose.
|
|
//
|
|
// IT TAKES THE WHOLE CONFIG rather than the one field it needs, and that is
|
|
// the same CONVE-19 argument one level along (BUG-2736). The phase-2 flip
|
|
// began as a hand-picked `cfg.EventsPublishEpoch` argument at the two call
|
|
// sites in RunE; replacing it with `false` there compiled, passed every test
|
|
// in the tree, and left the deployment silently stuck on phase 1 —
|
|
// indistinguishable from a correct phase-1 deployment, since phase 1 is the
|
|
// default. Reading the field HERE puts the link inside a function a test can
|
|
// call, and a caller that drops the config does not compile.
|
|
//
|
|
// The flip reaches the Redis bus only. The in-process bus has no wire and
|
|
// identifies its ID space by its own incarnation base instead (see
|
|
// internal/idspace), so it ignores the field.
|
|
func newObservedEventBus(cfg *config.Config, rc *redis.Client, redisKeys redisns.Keys, m *metrics.Metrics) events.EventBus {
|
|
if rc != nil {
|
|
bus := events.NewRedisBusWithKeys(rc, redisKeys, cfg.EventsPublishEpoch, cfg.EventsHeartbeat)
|
|
bus.SetObserver(metrics.NewEventsObserver(m))
|
|
return bus
|
|
}
|
|
bus := events.New()
|
|
bus.SetObserver(metrics.NewEventsObserver(m))
|
|
return bus
|
|
}
|