Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
49 commits
Select commit Hold shift + click to select a range
9716ef4
Take the account event feed into a durable ledger
jorgemanrubia Sep 16, 2026
e4852c2
Keep intake's recovery paths from swallowing, stalling or skipping
jorgemanrubia Sep 16, 2026
076b24e
Stop a repair walk from walking into the same 410 on every pass
jorgemanrubia Sep 16, 2026
a8e461b
Keep the ledger private, refuse feed redirects, and hand off through …
jorgemanrubia Sep 16, 2026
224de40
Make the adapter the only policy boundary, and state the repair pass
jorgemanrubia Sep 16, 2026
99359c0
Re-enter a refused position safely even from empty pages, and serve n…
jorgemanrubia Sep 16, 2026
e1a91fc
Retry the opening migration while SQLite reports the ledger busy
jorgemanrubia Sep 16, 2026
b7cc7cf
Count what the poll lane served, and replay rather than skip on a fil…
jorgemanrubia Sep 16, 2026
f58d62f
Let the replay label cover only the replay
jorgemanrubia Sep 16, 2026
4bf12ba
Deliver queue edges in order, report partial reconciliation honestly,…
jorgemanrubia Sep 16, 2026
3279414
Store ledger timestamps at fixed width, so retention compares them in…
jorgemanrubia Sep 16, 2026
8226ec9
Test ledger retention in both directions across fractional-second widths
jorgemanrubia Sep 16, 2026
7258711
Take the SDK's live feed seams now that they have landed
jorgemanrubia Sep 16, 2026
81b4797
Re-enter a restart at this filter set's own id, stop on a refused re-…
jorgemanrubia Sep 16, 2026
ae21cca
Bump the SDK to main for its live feed seams, and re-sync the model a…
jorgemanrubia Sep 16, 2026
f999fbb
Regenerate the vendored model on main's policy
jorgemanrubia Sep 16, 2026
94a9d1e
Restore the SQLite driver to go.mod and refresh the vendorHash on mai…
jorgemanrubia Sep 16, 2026
de87973
Strip terminal controls from pointer lines, and share one whole-line …
jorgemanrubia Sep 16, 2026
b592064
Keep an explicit --since as the entry until a checkpoint supersedes it
jorgemanrubia Sep 16, 2026
00be65c
Bound repair passes against cycles and endless pages, propagate cance…
jorgemanrubia Sep 16, 2026
3d93b57
Re-apply this branch's requirements on main's dependency bump
jorgemanrubia Sep 16, 2026
cd3c9b8
One NDJSON writer per sink, and an unknown membership snapshot that s…
jorgemanrubia Sep 16, 2026
0af1290
Repair a loss under the filters it was recorded with, and keep learne…
jorgemanrubia Sep 16, 2026
c9f54c2
Key only reference-like sinks in the line-writer registry
jorgemanrubia Sep 16, 2026
4228fa2
Tell a loss with no filters from a loss written before filters were s…
jorgemanrubia Sep 16, 2026
ce1ffa8
Own the filter set, bound repair fan-out, refuse an in-memory ledger,…
jorgemanrubia Sep 16, 2026
4bcd8ef
Fix the repair pool's shutdown, and make a learned project revocable
jorgemanrubia Sep 16, 2026
b2aad2f
Give each Run its own repair pool
jorgemanrubia Sep 16, 2026
38f439f
Hand over every record still unjudged on a start
jorgemanrubia Sep 16, 2026
f4c4c8a
Say that a re-offered record's pointer line is not re-emitted
jorgemanrubia Sep 16, 2026
0670c70
Derive a queue crossing from the operation that caused it, not a late…
jorgemanrubia Sep 17, 2026
b76da16
Wait as long as a throttled repair poll was told to
jorgemanrubia Sep 17, 2026
251aa9e
A throttle is not a verdict, and a new Run starts clean
jorgemanrubia Sep 17, 2026
886ec7d
Never strand a loss, and never let a throttle spend its window
jorgemanrubia Sep 17, 2026
329be35
Pin the in-window postpone on its own
jorgemanrubia Sep 17, 2026
458c064
Use the walk's own context in the postpone test
jorgemanrubia Sep 17, 2026
753affa
Close the throttle, the pool's bookkeeping, and the stdout stream's l…
jorgemanrubia Sep 17, 2026
e8a0d4f
Report an absolute-deadline close for what it was, and unlatch the ed…
jorgemanrubia Sep 17, 2026
8e1e98c
Pair the edge drain's unlock with the lock it takes
jorgemanrubia Sep 17, 2026
a10a38c
Make the instance lock a lock, and the pause edges ordered
jorgemanrubia Sep 17, 2026
20eba38
Validate the ledger this process opens, and let no committed event stop
jorgemanrubia Sep 17, 2026
f022c1a
Report the create that fails, and forget only what was offered
jorgemanrubia Sep 17, 2026
3963d28
Make terminal mean terminal
jorgemanrubia Sep 17, 2026
9c89aca
Let a refused position die with the epoch that refused it
jorgemanrubia Sep 17, 2026
f427ce7
Build the verdict stream's writer before the workers exist
jorgemanrubia Sep 17, 2026
0d84ab6
Keep a server's patience inside the deadline it is asking against
jorgemanrubia Sep 17, 2026
90e644a
Let a panicking callback cost the report, not the queue
jorgemanrubia Sep 17, 2026
5e448b8
Refuse a filter this lane cannot honor, and say where a panic went
jorgemanrubia Sep 17, 2026
d13fda8
Adopt the queue's logger under the queue's lock
jorgemanrubia Sep 17, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ require (
charm.land/bubbles/v2 v2.2.1
charm.land/bubbletea/v2 v2.0.9
charm.land/lipgloss/v2 v2.0.6
github.com/basecamp/basecamp-sdk/go v0.18.1-0.20260916210227-4523eac74cfa
github.com/basecamp/basecamp-sdk/go v0.19.0
github.com/basecamp/cli v0.2.2-0.20260828230226-767413fc712d
github.com/basecamp/mcp v0.0.0-20260828100356-2d6f44b51e9d
github.com/basecamp/surfguard/go v0.1.0
Expand All @@ -30,6 +30,7 @@ require (
golang.org/x/sys v0.48.0
golang.org/x/text v0.42.0
gopkg.in/yaml.v3 v3.0.1
modernc.org/sqlite v1.59.0
)

require (
Expand All @@ -54,6 +55,7 @@ require (
github.com/charmbracelet/x/windows v0.2.2 // indirect
github.com/clipperhouse/displaywidth v0.11.0 // indirect
github.com/clipperhouse/uax29/v2 v2.7.0 // indirect
github.com/coder/websocket v1.8.15 // indirect
github.com/cyberphone/json-canonicalization v0.0.0-20241213102144-19d51d7fe467 // indirect
github.com/danieljoos/wincred v1.2.3 // indirect
github.com/digitorus/pkcs7 v0.0.0-20230818184609-3a137a874352 // indirect
Expand Down Expand Up @@ -99,7 +101,7 @@ require (
github.com/inconshreveable/mousetrap v1.1.0 // indirect
github.com/itchyny/timefmt-go v0.1.8 // indirect
github.com/lucasb-eyer/go-colorful v1.4.1 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/mattn/go-isatty v0.0.24 // indirect
github.com/mattn/go-localereader v0.0.1 // indirect
github.com/mattn/go-runewidth v0.0.27 // indirect
github.com/microcosm-cc/bluemonday v1.0.27 // indirect
Expand All @@ -108,10 +110,12 @@ require (
github.com/muesli/cancelreader v0.2.2 // indirect
github.com/muesli/reflow v0.3.0 // indirect
github.com/muesli/termenv v0.16.0 // indirect
github.com/ncruces/go-strftime v1.0.0 // indirect
github.com/oapi-codegen/runtime v1.7.0 // indirect
github.com/oklog/ulid/v2 v2.1.1 // indirect
github.com/opencontainers/go-digest v1.0.0 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
github.com/rivo/uniseg v0.4.7 // indirect
github.com/secure-systems-lab/go-securesystemslib v0.11.0 // indirect
github.com/segmentio/asm v1.1.3 // indirect
Expand Down Expand Up @@ -144,4 +148,7 @@ require (
google.golang.org/grpc v1.83.2 // indirect
google.golang.org/protobuf v1.36.11 // indirect
k8s.io/klog/v2 v2.140.0 // indirect
modernc.org/libc v1.75.7 // indirect
modernc.org/mathutil v1.7.1 // indirect
modernc.org/memory v1.12.1 // indirect
)
47 changes: 42 additions & 5 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -87,8 +87,8 @@ github.com/aymanbagabas/go-udiff v0.4.1 h1:OEIrQ8maEeDBXQDoGCbbTTXYJMYRCRO1fnodZ
github.com/aymanbagabas/go-udiff v0.4.1/go.mod h1:0L9PGwj20lrtmEMeyw4WKJ/TMyDtvAoK9bf2u/mNo3w=
github.com/aymerick/douceur v0.2.0 h1:Mv+mAeH1Q+n9Fr+oyamOlAkUNPWPlA8PPGR0QAaYuPk=
github.com/aymerick/douceur v0.2.0/go.mod h1:wlT5vV2O3h55X9m7iVYN0TBM0NH/MmbLnd30/FjWUq4=
github.com/basecamp/basecamp-sdk/go v0.18.1-0.20260916210227-4523eac74cfa h1:efZJJSiwKn5lBih05oVfLM6S+ZYN6kB0vcH1DQmd8a8=
github.com/basecamp/basecamp-sdk/go v0.18.1-0.20260916210227-4523eac74cfa/go.mod h1:kIBDYwPMMD59PadNGxpH0YTQuI+blFPZ8MelGI0RK5Q=
github.com/basecamp/basecamp-sdk/go v0.19.0 h1:byygVVbJnWCZsyBNeAlztlUAV23ytLEhPx98WakNy+c=
github.com/basecamp/basecamp-sdk/go v0.19.0/go.mod h1:kIBDYwPMMD59PadNGxpH0YTQuI+blFPZ8MelGI0RK5Q=
github.com/basecamp/cli v0.2.2-0.20260828230226-767413fc712d h1:jAzDrCCzDpIwhbFT1xVVs0z2xpXoDEkomHfKB2bUUp8=
github.com/basecamp/cli v0.2.2-0.20260828230226-767413fc712d/go.mod h1:iTBTaWvsPEFIcZfkxQHEfISyJ6sZ7036K6bNx0RY3EE=
github.com/basecamp/mcp v0.0.0-20260828100356-2d6f44b51e9d h1:zEQVGq1x1nhKMZ2TudFAcSJ32CHT8richI1vQakIKz4=
Expand Down Expand Up @@ -148,6 +148,8 @@ github.com/clipperhouse/uax29/v2 v2.7.0 h1:+gs4oBZ2gPfVrKPthwbMzWZDaAFPGYK72F0NJ
github.com/clipperhouse/uax29/v2 v2.7.0/go.mod h1:EFJ2TJMRUaplDxHKj1qAEhCtQPW2tJSwu5BF98AuoVM=
github.com/codahale/rfc6979 v0.0.0-20141003034818-6a90f24967eb h1:EDmT6Q9Zs+SbUoc7Ik9EfrFqcylYqgPZ9ANSbTAntnE=
github.com/codahale/rfc6979 v0.0.0-20141003034818-6a90f24967eb/go.mod h1:ZjrT6AXHbDs86ZSdt/osfBi5qfexBrKUdONk989Wnk4=
github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
github.com/coreos/go-oidc/v3 v3.17.0 h1:hWBGaQfbi0iVviX4ibC7bk8OKT5qNr4klBaCHVNvehc=
github.com/coreos/go-oidc/v3 v3.17.0/go.mod h1:wqPbKFrVnE90vty060SB40FCJ8fTHTxSwyXJqZH+sI8=
github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
Expand Down Expand Up @@ -254,6 +256,8 @@ github.com/google/go-containerregistry v0.21.7 h1:/vPFuVXDjtFREsVArW+0h1CIl5urnO
github.com/google/go-containerregistry v0.21.7/go.mod h1:kjSbt7/zMsKLWfnHrIvKvhXHUw91jbe9DNjPPJ32gXE=
github.com/google/jsonschema-go v0.4.3 h1:/DBOLZTfDow7pe2GmaJNhltueGTtDKICi8V8p+DQPd0=
github.com/google/jsonschema-go v0.4.3/go.mod h1:r5quNTdLOYEz95Ru18zA0ydNbBuYoo9tgaYcxEYhJVE=
github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3 h1:LMLX+LgTNWpfvCBdFebv6EsYotImrt/Ppc5cXIriCSo=
github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3/go.mod h1:jl5iWTm0/hd5PjEYEOuwAJ57L/CibdZfrqZ5XA5GrCk=
github.com/google/s2a-go v0.1.9 h1:LGD7gtMgezd8a/Xak7mEWL0PjoTQFvpRudN895yqKW0=
github.com/google/s2a-go v0.1.9/go.mod h1:YA0Ei2ZQL3acow2O62kdp9UlnvMmU7kA6Eutn0dXayM=
github.com/google/trillian v1.7.3 h1:hziW+vo4czis48tzx2GK5xRBl/ZxBA9B0/UR5avXOro=
Expand Down Expand Up @@ -286,6 +290,8 @@ github.com/hashicorp/go-secure-stdlib/strutil v0.1.2 h1:kes8mmyCpxJsI7FTwtzRqEy9
github.com/hashicorp/go-secure-stdlib/strutil v0.1.2/go.mod h1:Gou2R9+il93BqX25LAKCLuM+y9U2T4hlwvT1yprcna4=
github.com/hashicorp/go-sockaddr v1.0.7 h1:G+pTkSO01HpR5qCxg7lxfsFEZaG+C0VssTy/9dbT+Fw=
github.com/hashicorp/go-sockaddr v1.0.7/go.mod h1:FZQbEYa1pxkQ7WLpyXJ6cbjpT8q0YgQaK/JakXqGyWw=
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
github.com/hashicorp/hcl v1.0.1-vault-7 h1:ag5OxFVy3QYTFTJODRzTKVZ6xvdfLLCA1cy/Y6xGI0I=
github.com/hashicorp/hcl v1.0.1-vault-7/go.mod h1:XYhtn6ijBSAj6n4YqAaf7RBPS4I06AItNorpy+MoQNM=
github.com/hashicorp/vault/api v1.22.0 h1:+HYFquE35/B74fHoIeXlZIP2YADVboaPjaSicHEZiH0=
Expand Down Expand Up @@ -319,8 +325,8 @@ github.com/letsencrypt/boulder v0.20260309.0 h1:kZynrxK3QfqLGx6hhoz+Rfs3hgltJs1p
github.com/letsencrypt/boulder v0.20260309.0/go.mod h1:yG8lj8pNPZ8taq3oNdTpfBS+eC74IaEuiewqzVpXiWE=
github.com/lucasb-eyer/go-colorful v1.4.1 h1:1EO+WB73+EH8EVbzlrG3KLAfEypQWVHIBqlTf+2hNss=
github.com/lucasb-eyer/go-colorful v1.4.1/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI=
github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A=
github.com/mattn/go-localereader v0.0.1 h1:ygSAOl7ZXTx4RdPYinUpg6W99U8jWvWi9Ye2JC/oIi4=
github.com/mattn/go-localereader v0.0.1/go.mod h1:8fBrzywKY7BI3czFoHkuzRoWE9C+EiG4R1k4Cjx5p88=
github.com/mattn/go-runewidth v0.0.12/go.mod h1:RAqKPSqVFrSLVXbA8x7dzmKdmGzieGRCM46jaSJTDAk=
Expand All @@ -346,6 +352,8 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc
github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk=
github.com/natefinch/atomic v1.0.1 h1:ZPYKxkqQOx3KZ+RsbnP/YsgvxWQPGxjC0oBt2AhwV0A=
github.com/natefinch/atomic v1.0.1/go.mod h1:N/D/ELrljoqDyT3rZrsUmtsuzvHkeB/wWjHV22AZRbM=
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
github.com/oapi-codegen/nullable v1.1.0 h1:eAh8JVc5430VtYVnq00Hrbpag9PFRGWLjxR1/3KntMs=
github.com/oapi-codegen/nullable v1.1.0/go.mod h1:KUZ3vUzkmEKY90ksAmit2+5juDIhIZhfDl+0PwOQlFY=
github.com/oapi-codegen/runtime v1.7.0 h1:t7358VYPvNbWJ9gdAkIK/smVeHpBf6yp8VTsaZsb/7k=
Expand All @@ -360,6 +368,8 @@ github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjL
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
Expand Down Expand Up @@ -487,7 +497,6 @@ golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7
golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk=
golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0=
golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE=
Expand Down Expand Up @@ -519,6 +528,34 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
k8s.io/klog/v2 v2.140.0 h1:Tf+J3AH7xnUzZyVVXhTgGhEKnFqye14aadWv7bzXdzc=
k8s.io/klog/v2 v2.140.0/go.mod h1:o+/RWfJ6PwpnFn7OyAG3QnO47BFsymfEfrz6XyYSSp0=
modernc.org/cc/v4 v4.29.2 h1:h6+9ciCnPKutf4I03CvheAvDLX7+IHlqR6Iy6J+cgd8=
modernc.org/cc/v4 v4.29.2/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI=
modernc.org/ccgo/v4 v4.35.0 h1:F+TUsmw09QxLzmi3aeYYGxjAXarmZaKgj3mKQHNaA8w=
modernc.org/ccgo/v4 v4.35.0/go.mod h1:qrVGs9S3Sr2Ztcg9ve+kTAYMp5a3YvWjo+SoN06kJ5I=
modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM=
modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU=
modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI=
modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
modernc.org/gc/v3 v3.1.5 h1:21ldfPfRYE31Tb7B3mwAK8gy1AxP4+dKjrOQPfqakoc=
modernc.org/gc/v3 v3.1.5/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY=
modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks=
modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI=
modernc.org/libc v1.75.7 h1:o3DTP9/0p9pKmY2WCKQaySW6wIiZhNM7wc2lUoyhfew=
modernc.org/libc v1.75.7/go.mod h1:bO5o2ztHxBb2rjz0PgdHN0sSMw57CgxGFLZ3Qd/QpVQ=
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
modernc.org/memory v1.12.1 h1:nFMiWrpStgZczNl6XI9GnIk/rWhYIyHGUaR04pGbp9g=
modernc.org/memory v1.12.1/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg=
modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
modernc.org/sqlite v1.59.0 h1:X1es1GpqBlS/5T+vbM4HLUdaa8OtQx468DF2vrx+38A=
modernc.org/sqlite v1.59.0/go.mod h1:+paeT2A3iPRHkQDwG7oA6Tk0zQd5woMEI8q7orfry8k=
modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs=
sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4=
software.sslmate.com/src/go-pkcs12 v0.4.0 h1:H2g08FrTvSFKUj+D309j1DPfk5APnIdAQAB8aEykJ5k=
Expand Down
66 changes: 62 additions & 4 deletions internal/connector/admission/commit_test.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
package admission

import (
"bufio"
"bytes"
"context"
"encoding/json"
"errors"
Expand Down Expand Up @@ -396,15 +398,15 @@ func (stuckWriter) Write([]byte) (int, error) { return 0, nil }

func TestALineIsWrittenWholeOrNotAtAll(t *testing.T) {
w := &chunkWriter{chunk: 7}
require.NoError(t, (&lineWriter{w: w}).write(Verdict{EventID: 1, EventType: "card.created", State: StateDiscarded, Reason: ReasonNotInMatrix}))
require.NoError(t, newLineWriter(nil, w).write(Verdict{EventID: 1, EventType: "card.created", State: StateDiscarded, Reason: ReasonNotInMatrix}))
var m map[string]any
require.NoError(t, json.Unmarshal([]byte(strings.TrimSuffix(w.b.String(), "\n")), &m), "a writer that takes a line in pieces still gets all of it")
assert.True(t, strings.HasSuffix(w.b.String(), "}\n"))

// Bounded: a writer loop that forgot this case would spin, and should fail
// this test by name rather than hang the suite.
done := make(chan error, 1)
go func() { done <- (&lineWriter{w: stuckWriter{}}).write(Verdict{EventID: 1, State: StateDiscarded}) }()
go func() { done <- newLineWriter(nil, stuckWriter{}).write(Verdict{EventID: 1, State: StateDiscarded}) }()
select {
case err := <-done:
require.ErrorIs(t, err, io.ErrShortWrite)
Expand All @@ -416,7 +418,7 @@ func TestALineIsWrittenWholeOrNotAtAll(t *testing.T) {
func TestLinesCannotCarryTerminalControls(t *testing.T) {
esc, csi, bel := string(rune(0x1b)), string(rune(0x9b)), string(rune(0x07))
var b strings.Builder
require.NoError(t, (&lineWriter{w: &b}).write(Verdict{
require.NoError(t, newLineWriter(nil, &b).write(Verdict{
EventID: 1,
EventType: "card.created" + esc + "]0;owned" + bel,
RecordingURL: "https://app.basecamp.com/x" + csi + "31m" + esc + "[2J",
Expand Down Expand Up @@ -462,9 +464,65 @@ func TestRunStopsOnALoadFailure(t *testing.T) {

func TestLinesKeepLocalPathsAsWritten(t *testing.T) {
var b strings.Builder
require.NoError(t, (&lineWriter{w: &b}).write(Verdict{EventID: 1, State: StateAdmitted, Route: " /work/My Projects\tA", Class: "in ternal"}))
require.NoError(t, newLineWriter(nil, &b).write(Verdict{EventID: 1, State: StateAdmitted, Route: " /work/My Projects\tA", Class: "in ternal"}))
var m map[string]any
require.NoError(t, json.Unmarshal([]byte(b.String()), &m))
assert.Equal(t, " /work/My Projects\tA", m["route"], "whitespace in a path is not a terminal control")
assert.Equal(t, "in ternal", m["class"])
}

// A sink that is safe for concurrent use, so the only race a run can report is
// one of admission's own making.
type lockedSink struct {
mu sync.Mutex
buf bytes.Buffer
}

func (s *lockedSink) Write(p []byte) (int, error) {
s.mu.Lock()
defer s.mu.Unlock()
return s.buf.Write(p)
}

// The verdict writer is built before any worker exists. Built on first use, a
// pool of workers reaching their first lines together would read and assign it
// at once — a data race on the ordinary path, not an edge case.
func TestTheVerdictWriterIsBuiltBeforeAnyWorkerWrites(t *testing.T) {
sink := &lockedSink{}
lines := newLineWriter(nil, sink)

var wg sync.WaitGroup
for worker := range 8 {
wg.Go(func() {
for i := range 25 {
require.NoError(t, lines.write(Verdict{
EventID: int64(worker*100 + i),
EventType: "comment.created",
State: StateAdmitted,
}))
}
})
}
wg.Wait()

scanner := bufio.NewScanner(bytes.NewReader(sink.buf.Bytes()))
count := 0
for scanner.Scan() {
var line map[string]any
require.NoError(t, json.Unmarshal(scanner.Bytes(), &line), "torn line %d: %q", count, scanner.Text())
count++
}
require.NoError(t, scanner.Err())
assert.Equal(t, 200, count)
}

// One sink has one writer, whoever asks for it: the lock that keeps lines
// whole is only one lock if there is only one of it.
func TestOneSinkHasOneWriter(t *testing.T) {
sink := &lockedSink{}

first := newLineWriter(nil, sink)
second := newLineWriter(nil, sink)

assert.Same(t, first.out, second.out)
}
58 changes: 33 additions & 25 deletions internal/connector/admission/run.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,13 @@ package admission

import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"sync"

"github.com/basecamp/basecamp-cli/internal/connector/ndjson"
"github.com/basecamp/basecamp-cli/internal/richtext"
)

Expand Down Expand Up @@ -42,8 +42,12 @@ type RunOptions struct {
// Workers is the fetcher pool size; DefaultWorkers when zero.
Workers int
// Lines receives one NDJSON line per committed verdict.
Lines io.Writer
Logger *slog.Logger
Lines io.Writer
// LineWriter, when set, is the writer the lines go through, and Lines is
// ignored: one sink has one writer and one lock, and a caller wiring
// admission beside intake passes the same writer to both.
LineWriter *ndjson.Writer
Logger *slog.Logger
}

// Run takes ids until ctx ends, deciding and committing each. It returns nil
Expand All @@ -68,7 +72,9 @@ func Run(ctx context.Context, opts RunOptions) error {
if log == nil {
log = slog.New(slog.DiscardHandler)
}
lines := &lineWriter{w: opts.Lines}
// Built here, before the workers exist, so nothing is initialized on a
// path two of them can take at once.
lines := newLineWriter(opts.LineWriter, opts.Lines)

ctx, cancel := context.WithCancel(ctx)
defer cancel()
Expand Down Expand Up @@ -178,33 +184,35 @@ func LineFor(v Verdict) Line {
}
}

// lineWriter is the verdict stream. It holds a writer and nothing else: no
// flag, no once, no first-use path.
//
// It used to build the writer on the first verdict, which is a race the moment
// admission has more than one worker — two of them reaching their first line
// together read and assign the same field. Laziness is not a thing to guard
// here; it is a thing to remove. The writer is built once, before any worker
// starts, and injected.
type lineWriter struct {
mu sync.Mutex
w io.Writer
out *ndjson.Writer
}

// newLineWriter takes the writer a caller wiring admission beside intake
// passes in, or builds one for the sink. There is exactly one writer per sink
// either way: NewWriter returns the same writer for the same sink, so the one
// lock is the one lock.
func newLineWriter(out *ndjson.Writer, sink io.Writer) *lineWriter {
if out == nil && sink != nil {
out = ndjson.NewWriter(sink)
}
return &lineWriter{out: out}
}

func (l *lineWriter) write(v Verdict) error {
if l.w == nil {
if l.out == nil {
return nil
}
b, err := json.Marshal(LineFor(v))
if err != nil {
return fmt.Errorf("admission: encode line: %w", err)
}
l.mu.Lock()
defer l.mu.Unlock()
// One line, whole: a writer that takes part of it is written the rest,
// and one that takes none without an error has failed. A torn line is
// worse than no line to whoever parses the stream.
for rest := append(b, '\n'); len(rest) > 0; {
n, err := l.w.Write(rest)
if err != nil {
return fmt.Errorf("admission: write line: %w", err)
}
if n <= 0 {
return fmt.Errorf("admission: write line: %w", io.ErrShortWrite)
}
rest = rest[n:]
if err := l.out.WriteLine(LineFor(v)); err != nil {
return fmt.Errorf("admission: %w", err)
}
return nil
}
Loading
Loading