feat(auth): coordinate canonical projection publication
This commit is contained in:
@@ -3,12 +3,14 @@
|
||||
package authprojection
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sys/unix"
|
||||
)
|
||||
@@ -71,14 +73,24 @@ func Inspect(spec Spec) (Status, error) {
|
||||
}
|
||||
|
||||
func Begin(spec Spec, before *Snapshot, requireReadyMatch bool) (*Transaction, error) {
|
||||
return BeginContext(context.Background(), spec, before, requireReadyMatch)
|
||||
}
|
||||
|
||||
func BeginContext(ctx context.Context, spec Spec, before *Snapshot, requireReadyMatch bool) (*Transaction, error) {
|
||||
if ctx == nil {
|
||||
return nil, ErrIntegrity
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
root, err := openRoot(spec)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rendezvousBeforeFlock(root)
|
||||
if unix.Flock(root, unix.LOCK_EX) != nil {
|
||||
if err := flockContext(ctx, root); err != nil {
|
||||
unix.Close(root)
|
||||
return nil, ErrIntegrity
|
||||
return nil, err
|
||||
}
|
||||
transaction := &Transaction{spec: spec, before: before, rootFD: root, lockFD: root}
|
||||
selector, selectorErr := readSelector(root, spec)
|
||||
@@ -132,6 +144,30 @@ func Begin(spec Spec, before *Snapshot, requireReadyMatch bool) (*Transaction, e
|
||||
return transaction, nil
|
||||
}
|
||||
|
||||
func flockContext(ctx context.Context, descriptor int) error {
|
||||
for {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
err := unix.Flock(descriptor, unix.LOCK_EX|unix.LOCK_NB)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if !errors.Is(err, unix.EWOULDBLOCK) {
|
||||
return ErrIntegrity
|
||||
}
|
||||
timer := time.NewTimer(10 * time.Millisecond)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
if !timer.Stop() {
|
||||
<-timer.C
|
||||
}
|
||||
return ctx.Err()
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (transaction *Transaction) Commit(after Snapshot) (Status, error) {
|
||||
if transaction == nil || transaction.closed || !transaction.blocked || validateSnapshot(after) != nil {
|
||||
return Status{}, ErrIntegrity
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
package authprojection
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
@@ -11,10 +12,50 @@ import (
|
||||
"sync"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sys/unix"
|
||||
)
|
||||
|
||||
func TestBeginContextReturnsWhenRuntimeFlockIsHeld(t *testing.T) {
|
||||
spec := testSpec(t)
|
||||
first, err := Begin(spec, nil, false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = first.Close() })
|
||||
contended := make(chan error, 1)
|
||||
restore := setTestHooksForTest(testHooks{beforeFlock: func(fd int) {
|
||||
if err := unix.Flock(fd, unix.LOCK_EX|unix.LOCK_NB); !errors.Is(err, unix.EWOULDBLOCK) {
|
||||
contended <- fmt.Errorf("runtime lock contention = %v", err)
|
||||
return
|
||||
}
|
||||
contended <- nil
|
||||
}})
|
||||
t.Cleanup(restore)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
result := make(chan error, 1)
|
||||
go func() {
|
||||
transaction, err := BeginContext(ctx, spec, nil, false)
|
||||
if err == nil {
|
||||
err = transaction.Close()
|
||||
}
|
||||
result <- err
|
||||
}()
|
||||
if err := <-contended; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cancel()
|
||||
select {
|
||||
case err := <-result:
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("BeginContext() error = %v, want context cancellation", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("BeginContext() did not return after cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
func testSpec(t *testing.T) Spec {
|
||||
t.Helper()
|
||||
root := t.TempDir()
|
||||
|
||||
@@ -2,8 +2,15 @@
|
||||
|
||||
package authprojection
|
||||
|
||||
func Inspect(Spec) (Status, error) { return Status{}, ErrUnsupported }
|
||||
func Begin(Spec, *Snapshot, bool) (*Transaction, error) { return nil, ErrUnsupported }
|
||||
func (*Transaction) Commit(Snapshot) (Status, error) { return Status{}, ErrUnsupported }
|
||||
func (*Transaction) RestoreIfUnchanged(Snapshot) error { return ErrUnsupported }
|
||||
func (*Transaction) Close() error { return ErrUnsupported }
|
||||
import "context"
|
||||
|
||||
func Inspect(Spec) (Status, error) { return Status{}, ErrUnsupported }
|
||||
func Begin(Spec, *Snapshot, bool) (*Transaction, error) {
|
||||
return nil, ErrUnsupported
|
||||
}
|
||||
func BeginContext(context.Context, Spec, *Snapshot, bool) (*Transaction, error) {
|
||||
return nil, ErrUnsupported
|
||||
}
|
||||
func (*Transaction) Commit(Snapshot) (Status, error) { return Status{}, ErrUnsupported }
|
||||
func (*Transaction) RestoreIfUnchanged(Snapshot) error { return ErrUnsupported }
|
||||
func (*Transaction) Close() error { return ErrUnsupported }
|
||||
|
||||
Reference in New Issue
Block a user