amadan

oficina / home Public

Interrupt private request output when its context ends

Paul Campbell pushed by paul@keymail.dev 7e282e2a363b9cc90cd74dd97809e6613a699af8 parent caa16b5
2 files changed, +189 −1
  • internal/provisioner/prepare_linux.go +25 −1
  • internal/provisioner/prepare_linux_test.go +164 −0

CI passed — run details

Intent

full ledger: feat-provisioner-prepare

diff --git a/internal/provisioner/prepare_linux.go b/internal/provisioner/prepare_linux.go
index 1e45a93..56de95e 100644
--- a/internal/provisioner/prepare_linux.go
+++ b/internal/provisioner/prepare_linux.go
@@ -8,6 +8,7 @@ import (
"net/mail"
"os"
"strconv"
+ "time"
"amadan.net/oficina/home/internal/operator"
"github.com/go-json-experiment/json"
@@ -62,6 +63,11 @@ func privateRequestOutput(fd int, excluded ...int) (*requestOutput, error) {
return nil, ErrRefused
}
}
+ // Pollable nonblocking ownership lets Close interrupt a blocked Write.
+ // This descriptor is explicitly transferred to the native command.
+ if unix.SetNonblock(fd, true) != nil {
+ return nil, ErrRefused
+ }
file := os.NewFile(uintptr(fd), "private-request")
if file == nil {
return nil, ErrRefused
@@ -154,7 +160,7 @@ func prepareRequest(ctx context.Context, statePath, floorPath string, password,
}
// The record and independent floor are durable before output. Failure cannot
// release its role slot or cause the next call to sign a replacement.
- if e = writePrivateRequest(output.file, raw); e != nil {
+ if e = output.write(ctx, raw); e != nil {
return PreparationInfo{}, e
}
return PreparationInfo{p.Installation, p.Operation, p.Role, req.Authorization().Sequence, "local-request"}, nil
@@ -169,3 +175,21 @@ func writePrivateRequest(output io.Writer, raw []byte) error {
}
return nil
}
+
+// write keeps a stalled recipient from retaining the journal's process lock.
+// Closing a pollable pipe interrupts its wait without discarding the saved
+// request. The recipient may have a prefix, so a retry still uses exact bytes.
+func (o *requestOutput) write(ctx context.Context, raw []byte) error {
+ if o == nil || o.file == nil || len(raw) == 0 || len(raw) > operator.BodyLimit {
+ return ErrRefused
+ }
+ bounded, cancel := context.WithTimeout(ctx, 30*time.Second)
+ defer cancel()
+ stop := context.AfterFunc(bounded, func() { o.Close() })
+ defer stop()
+ e := writePrivateRequest(o.file, raw)
+ if bounded.Err() != nil {
+ return ErrUnavailable
+ }
+ return e
+}
diff --git a/internal/provisioner/prepare_linux_test.go b/internal/provisioner/prepare_linux_test.go
index 06ca17f..ee268a9 100644
--- a/internal/provisioner/prepare_linux_test.go
+++ b/internal/provisioner/prepare_linux_test.go
@@ -4,11 +4,14 @@ package provisioner
import (
"bytes"
+ "context"
+ "database/sql"
"io"
"os"
"os/exec"
"strconv"
"testing"
+ "time"
"amadan.net/oficina/home/internal/operator"
"github.com/go-json-experiment/json"
@@ -322,3 +325,164 @@ func TestPreparationClosedPlanRefusesUnknownAndDuplicateFields(t *testing.T) {
}
}
}
+
+func TestFullPrivatePipeCancellationKeepsExactRequest(t *testing.T) {
+ j, p := stateFixture(t)
+ op := testToken(23)
+ raw, e := j.prepare(context.Background(), p.Installation, op, "directory.initialize", initialization(op))
+ if e != nil {
+ t.Fatal(e)
+ }
+ fds := make([]int, 2)
+ if e = unix.Pipe2(fds, unix.O_CLOEXEC|unix.O_NONBLOCK); e != nil {
+ t.Fatal(e)
+ }
+ defer unix.Close(fds[0])
+ output, e := privateRequestOutput(fds[1])
+ if e != nil {
+ unix.Close(fds[1])
+ t.Fatal(e)
+ }
+ defer output.Close()
+ fill := bytes.Repeat([]byte("x"), 4096)
+ filled := 0
+ for {
+ n, err := unix.Write(fds[1], fill)
+ filled += n
+ if err == unix.EAGAIN {
+ break
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ if filled <= 0 {
+ t.Fatal("pipe was not filled")
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { done <- output.write(ctx, raw) }()
+ select {
+ case e := <-done:
+ t.Fatal("full pipe did not block", e)
+ case <-time.After(50 * time.Millisecond):
+ }
+ cancel()
+ select {
+ case e := <-done:
+ if e != ErrUnavailable {
+ t.Fatal("cancellation error", e)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("cancellation did not interrupt pipe")
+ }
+ j.close()
+ reopened, e := openJournal(j.dir, j.floor, j.key, false)
+ if e != nil {
+ t.Fatal("process lock retained", e)
+ }
+ defer reopened.close()
+ repeated, e := reopened.prepare(context.Background(), p.Installation, op, "directory.initialize", initialization(op))
+ if e != nil || !bytes.Equal(raw, repeated) {
+ t.Fatal("interrupted output lost exact request", e)
+ }
+ s := snapshot(t, reopened).installations[p.Installation]
+ if s.Operations[op].Status != "prepared" || s.Heads["directory"].Sequence != 0 {
+ t.Fatal("cancellation implied execution")
+ }
+}
+
+func TestPrepareRequestCancelsFullPipeAndReleasesRegistry(t *testing.T) {
+ state, floor := persistentDirectory(t), persistentDirectory(t)
+ password := []byte("synthetic-full-pipe-password")
+ if _, e := Initialize(state, floor, password); e != nil {
+ t.Fatal(e)
+ }
+ enrollment, _ := canonicalJSON(planFixture())
+ if _, e := EnrollPlan(state, floor, password, enrollment); e != nil {
+ t.Fatal(e)
+ }
+ p := ownerOperationPlan()
+ plan, _ := json.Marshal(p, json.Deterministic(true))
+ fds := make([]int, 2)
+ if e := unix.Pipe2(fds, unix.O_CLOEXEC|unix.O_NONBLOCK); e != nil {
+ t.Fatal(e)
+ }
+ defer unix.Close(fds[0])
+ output, e := privateRequestOutput(fds[1])
+ if e != nil {
+ unix.Close(fds[1])
+ t.Fatal(e)
+ }
+ defer output.Close()
+ for {
+ _, e = unix.Write(fds[1], bytes.Repeat([]byte("x"), 4096))
+ if e == unix.EAGAIN {
+ break
+ }
+ if e != nil {
+ t.Fatal(e)
+ }
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { _, e := prepareRequest(ctx, state, floor, password, plan, output); done <- e }()
+ // Observe the immutable fixture record without joining its writer lock.
+ observer, e := sql.Open("sqlite", "file:"+state+"/registry.db?mode=ro")
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer observer.Close()
+ deadline := time.Now().Add(30 * time.Second)
+ for {
+ var n int
+ e = observer.QueryRow("SELECT count(*) FROM records WHERE operation=? AND state='finalized'", p.Operation).Scan(&n)
+ if e == nil && n == 1 {
+ break
+ }
+ select {
+ case e := <-done:
+ t.Fatal("preparation stopped before output", e)
+ default:
+ }
+ if time.Now().After(deadline) {
+ t.Fatal("fixture request was not persisted")
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ select {
+ case e := <-done:
+ t.Fatal("full output did not wait", e)
+ case <-time.After(50 * time.Millisecond):
+ }
+ cancel()
+ select {
+ case e := <-done:
+ if e != ErrUnavailable {
+ t.Fatal("canceled preparation", e)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("prepareRequest ignored cancellation")
+ }
+ observer.Close()
+ j, e := existingJournal(state, floor, password)
+ if e != nil {
+ t.Fatal("registry lock not released", e)
+ }
+ defer closeExisting(j)
+ s := snapshot(t, j).installations[p.Installation]
+ saved := s.Operations[p.Operation]
+ if saved == nil || saved.Status != "prepared" || s.Heads[p.Role].Sequence != 0 {
+ t.Fatal("canceled output changed authority")
+ }
+ _, input, e := decodeOperationPlan(plan)
+ if e != nil {
+ t.Fatal(e)
+ }
+ replay, e := j.prepare(context.Background(), p.Installation, p.Operation, p.Action, input)
+ if e != nil || !bytes.Equal(replay, saved.Raw) {
+ t.Fatal("canceled prepare lost exact retry", e)
+ }
+}