From 06c14a6837fb4255b4b9e186a0a2060fbee5ecff Mon Sep 17 00:00:00 2001 From: guangtao Date: Fri, 30 Oct 2026 23:35:54 +0800 Subject: [PATCH] fix(control): recover original publication before activation --- dev/proofs/lean/PublicationRecovery.lean | 15 +++ dev/proofs/quint/AuthorityPublication.qnt | 21 +++- dev/sysml/WeaverExecution.sysml | 8 +- .../59.03-provider-rpc-core.d2 | 2 +- .../59.03-provider-rpc-core.svg | 40 +++--- .../59.34-control-publication-recovery.org | 88 ++++++++++++++ go/dev/provider_control_test.go | 55 ++++++++- go/dev/publication_recovery_test.go | 27 +++- go/internal/controlplane/gateway.go | 2 +- go/internal/gatewayclient/publication.go | 23 ++++ .../publication_activation_test.go | 115 ++++++++++++++++++ 11 files changed, 362 insertions(+), 34 deletions(-) create mode 100644 docs/50-59-research-program/59-provider-rpc/59.34-control-publication-recovery.org create mode 100644 go/internal/gatewayclient/publication_activation_test.go diff --git a/dev/proofs/lean/PublicationRecovery.lean b/dev/proofs/lean/PublicationRecovery.lean index 331519fb..a73ddb35 100644 --- a/dev/proofs/lean/PublicationRecovery.lean +++ b/dev/proofs/lean/PublicationRecovery.lean @@ -81,6 +81,20 @@ theorem recovery_and_provider_accounting_composition [DecidableEq Key] ⟨run_preserves_original_decisions run id decision original, (run_preserves_accounting h run).2.1, (run_preserves_accounting h run).2.2⟩ +/- A source observation binds the frozen command, not the current policy head. + Native refinement must authenticate InspectPublication and validate its full + domain-separated digest. Missing/failed reads do not establish abort. -/ +def ExactObservation (s : State Key) (id : Key) (payload parent fence : Nat) (d : Decision) : Prop := + s.records id = some d ∧ d.payload = payload ∧ d.parent = parent ∧ d.requestedFence = fence + +theorem exact_observation_and_accounting_survive_recovery [DecidableEq Key] + {id : Key} {d : Decision} {payload parent fence : Nat} + (safe : Safe held calls s) (run : Run s t) + (observed : ExactObservation s id payload parent fence d) : + ExactObservation t id payload parent fence d ∧ t.held = held ∧ t.calls = calls := by + have preserved := recovery_and_provider_accounting_composition safe run observed.1 + exact ⟨⟨preserved.1, observed.2⟩, preserved.2⟩ + /- Owned-connection refinement: the worker has opened one private connection; every command is checked quiescent before/after SQL, storage faults terminate the owner, and a successful finish requires an explicitly successful close. @@ -161,4 +175,5 @@ theorem owned_worker_and_publication_composition [DecidableEq Key] #print axioms run_preserves_original_decisions #print axioms run_preserves_accounting #print axioms recovery_and_provider_accounting_composition +#print axioms exact_observation_and_accounting_survive_recovery end Weaver.PublicationRecovery diff --git a/dev/proofs/quint/AuthorityPublication.qnt b/dev/proofs/quint/AuthorityPublication.qnt index 14588e7a..c32bafb5 100644 --- a/dev/proofs/quint/AuthorityPublication.qnt +++ b/dev/proofs/quint/AuthorityPublication.qnt @@ -3,20 +3,24 @@ module AuthorityPublication { import PublicationRecovery as A from "./PublicationRecovery" // Go activation composes existing publication and execution models. It does // not redeclare their CAS, nonce, budget or Cedar semantics. - type Control={compiled:int,desired:int,registered:bool,ack:bool,active:bool} + type Control={compiled:int,desired:int,registered:bool,ack:bool,active:bool,submits:int} var c:Control val original=A::authority.records.get(0) val safe=E::safe and A::safe and + c.submits<=1 and (c.active implies c.compiled>=0 and c.registered and original.outcome==1 and original.payload==c.compiled and c.ack) and (A::authority.head>=0 implies E::s.revision==A::authority.head and E::s.fence==A::authority.fence) - action init=all {E::init,A::init,c'={compiled:-1,desired:0,registered:false,ack:false,active:false}} + action init=all {E::init,A::init,c'={compiled:-1,desired:0,registered:false,ack:false,active:false,submits:0}} // Concrete source admission is witnessed by Scheme/FFI/Go tests, not a // duplicate Boolean implementation of those validators in this model. action compile(accepted:bool):bool=all {accepted,c.compiled<0,c'=c.with("compiled",c.desired),E::observe,A::observe} action editSource=all {c'=c.with("desired",1),E::observe,A::observe} action register=all {c.compiled>=0,c'=c.with("registered",true),E::observe,A::observe} - action commit=all {c.registered, - any {A::publish(0,c.compiled,-1,1),A::replay(0,c.compiled,-1,1)},c'=c,E::observe} + // A known durable outcome is observed, never submitted again by bootstrap. + action commit=all {c.registered,original.outcome==0, + A::publish(0,c.compiled,-1,1),c'=c.with("submits",c.submits+1),E::observe} + action inspectMissing=all {c.compiled>=0,original.outcome==0,c'=c,E::observe,A::observe} + action inspectUnavailable=all {c'=c,E::observe,A::observe} action acknowledge=all {original.outcome==1,original.payload==c.compiled,c'=c.with("ack",true),E::observe,A::observe} action recover(source:int):bool=all {original.outcome==1,source==c.compiled,A::replay(0,source,-1,1),c'=c.with("ack",true),E::observe} action activate=all {original.outcome==1,c.ack,original.payload==c.compiled,c'=c.with("active",true),E::observe,A::observe} @@ -31,18 +35,25 @@ module AuthorityPublication { action invoke=all {c.active,c'=c,E::invoke,A::observe} action release=all {c.active,c'=c,E::batch(1,0),A::observe} action deliver=all {c'=c,E::deliver,A::observe} - action step=any {compile(true),editSource,register,commit,acknowledge,recover(0),recover(1),activate,loseReply,restart,publish,transport,admit,consume,burn,invoke,release,deliver} + action step=any {compile(true),editSource,register,commit,inspectMissing,inspectUnavailable,acknowledge,recover(0),recover(1),activate,loseReply,restart,publish,transport,admit,consume,burn,invoke,release,deliver} action prepared=init.then(compile(true)).then(register) action ready=prepared.then(commit).then(acknowledge).then(activate) action spent=ready.then(admit).then(consume).then(burn).then(invoke) action unsafeRecoverDesired=all {original.outcome==1,c'=c.with("compiled",c.desired).with("ack",true).with("active",true),E::observe,A::observe} action unsafeActivateWithoutOutcome=all {c'=c.with("active",true),E::observe,A::observe} action unsafeRetry=all {c'=c,E::unsafeRetryReset,A::observe} + action unnecessaryKnownOutcomeReplay=all {original.outcome==1,A::replay(0,c.compiled,-1,1),c'=c.with("submits",c.submits+1),E::observe} run authorityUncompiledSourceCannotRegisterTest=init.then(register.fail()) run authorityRejectedSourceCannotCompileTest=init.then(compile(false).fail()) run authoritySourceCommitNeverSendsTest=prepared.then(commit).expect(safe and E::s.calls==0 and E::s.held==0) run authorityLostReplyCannotActivateTest=prepared.then(commit).then(loseReply).then(activate.fail()) run authorityLostReplyAndOwnerRestartPreserveOutcomeTest=prepared.then(commit).then(loseReply).then(restart).then(recover(0)).then(activate).expect(safe and original.payload==0 and E::s.calls==0) + run authorityKnownOutcomeRecoveryDoesNotSubmitTest=ready.then(restart).then(recover(0)).then(activate).expect(safe and c.submits==1) + run authorityMissingInspectionDoesNotAbortLateCommitTest=prepared.then(inspectMissing).then(commit).then(recover(0)).then(activate).expect(safe and c.submits==1) + run authorityUnavailableInspectionDoesNotActivateTest=prepared.then(commit).then(loseReply).then(inspectUnavailable).expect(safe).then(activate.fail()) + // Rust idempotent replay remains safe. This witnesses a violation of the + // narrower Go bootstrap contract: recovery of a known decision must be read-only. + run authorityKnownOutcomeReplayContractCounterexampleTest=ready.then(unnecessaryKnownOutcomeReplay).expect(not(safe) and A::safe and E::safe) run authorityChangedCompiledSourceCannotRecoverTest=prepared.then(commit).then(editSource).then(recover(1).fail()) run authorityEditingFileCannotRewriteFrozenCommandTest=prepared.then(editSource).then(commit).then(acknowledge).then(activate).expect(safe and original.payload==0 and c.desired==1) run authorityHistoricalReceiptCannotOverrideRevocationTest=ready.then(publish).then(restart).then(recover(0)).then(activate).then(admit.fail()) diff --git a/dev/sysml/WeaverExecution.sysml b/dev/sysml/WeaverExecution.sysml index 1c74776b..10aa5b20 100644 --- a/dev/sysml/WeaverExecution.sysml +++ b/dev/sysml/WeaverExecution.sysml @@ -72,7 +72,13 @@ package WeaverExecution { action def Enqueue; action def Consume; action def GuardAndSend; - action def RecoverOutcome; + action def RecoverOutcome { + doc /* Go observes the Rust-owned durable result under the frozen + publication identity and digest before submitting an absent decision. + A known result is read-only recovery. Unknown or rejected results do + not activate ingress. A historical receipt never replaces current + Rust Policy, fence, original claim, history or budget checks. */ + } part gateway { part semanticPublisher : SchemeSemanticPublisher; diff --git a/docs/50-59-research-program/59-provider-rpc/59.03-provider-rpc-core.d2 b/docs/50-59-research-program/59-provider-rpc/59.03-provider-rpc-core.d2 index c3f475b5..566a5675 100644 --- a/docs/50-59-research-program/59-provider-rpc/59.03-provider-rpc-core.d2 +++ b/docs/50-59-research-program/59-provider-rpc/59.03-provider-rpc-core.d2 @@ -99,7 +99,7 @@ provider: LLM Provider / SSE\nOne consumed HTTPS invocation { height: 110 } source -> control: Governed config -control -> source_rpc: Publication +control -> source_rpc: Observe / initial CAS source_rpc -> owner: Policy / outcome CAS control -> rpc: Typed RPC rpc -> owner: Admit diff --git a/docs/50-59-research-program/59-provider-rpc/59.03-provider-rpc-core.svg b/docs/50-59-research-program/59-provider-rpc/59.03-provider-rpc-core.svg index 4973b36e..ac383c4d 100644 --- a/docs/50-59-research-program/59-provider-rpc/59.03-provider-rpc-core.svg +++ b/docs/50-59-research-program/59-provider-rpc/59.03-provider-rpc-core.svg @@ -1,23 +1,23 @@ -WEAVER / DEFAULT gRPC GATEWAYScheme / Yuanxu modulesGovernance Policy / source assetsBound source → Cedar FFISource mTLS gRPCAtomic task / profile registrationBounded publish / recovery laneTelemetry ServiceDefault local Go / Rust eventsPolicy-controlled exportProject SysML · Cedar · Quint · LeanGo Control PlaneSource-bound LLM routingDeadline-bound transport setupOperation mTLS gRPCPinned caller and method roleRust Deciding OwnerRegistry / claim / budget CASFresh Cedar dispatch and releaseGo Stream ConsumerOrdered data and canonical receiptOriginal Plan / Encoding TokenFresh Cedar Release GuardProtobuf or optional Arrow IPCLLM Provider / SSEOne consumed HTTPS invocation Governed configPublicationPolicy / outcome CASTyped RPCAdmitCedar + cancellation Bounded eventsSSEStream + .d2-3296100646 .fill-N1{fill:#0A0F25;} + .d2-3296100646 .fill-N2{fill:#676C7E;} + .d2-3296100646 .fill-N7{fill:#FFFFFF;} + .d2-3296100646 .fill-B1{fill:#000E3D;} + .d2-3296100646 .fill-B2{fill:#234CDA;} + .d2-3296100646 .fill-B6{fill:#E7EAFF;} + .d2-3296100646 .stroke-B1{stroke:#000E3D;} + .d2-3296100646 .stroke-B2{stroke:#234CDA;}.appendix text.text{fill:#0A0F25}.md{--color-fg-default:#0A0F25;--color-fg-muted:#676C7E;--color-fg-subtle:#9499AB;--color-canvas-default:#FFFFFF;--color-canvas-subtle:#EEF1F8;--color-border-default:#000E3D;--color-border-muted:#234CDA;--color-neutral-muted:#EEF1F8;--color-accent-fg:#234CDA;--color-accent-emphasis:#234CDA;--color-attention-subtle:#676C7E;--color-danger-fg:red;}.light-code{display: block}.dark-code{display: none}]]>WEAVER / DEFAULT gRPC GATEWAYScheme / Yuanxu modulesGovernance Policy / source assetsBound source → Cedar FFISource mTLS gRPCAtomic task / profile registrationBounded publish / recovery laneTelemetry ServiceDefault local Go / Rust eventsPolicy-controlled exportProject SysML · Cedar · Quint · LeanGo Control PlaneSource-bound LLM routingDeadline-bound transport setupOperation mTLS gRPCPinned caller and method roleRust Deciding OwnerRegistry / claim / budget CASFresh Cedar dispatch and releaseGo Stream ConsumerOrdered data and canonical receiptOriginal Plan / Encoding TokenFresh Cedar Release GuardProtobuf or optional Arrow IPCLLM Provider / SSEOne consumed HTTPS invocation Governed configObserve / initial CASPolicy / outcome CASTyped RPCAdmitCedar + cancellation Bounded eventsSSEStream - - + + diff --git a/docs/50-59-research-program/59-provider-rpc/59.34-control-publication-recovery.org b/docs/50-59-research-program/59-provider-rpc/59.34-control-publication-recovery.org new file mode 100644 index 00000000..f2846298 --- /dev/null +++ b/docs/50-59-research-program/59-provider-rpc/59.34-control-publication-recovery.org @@ -0,0 +1,88 @@ +#+title: 59.34 Control Activation from Original Publication Outcomes +#+language: en + +* Problem and resulting behavior +An authority commit may succeed while its gRPC response is lost. Previously, +Go bootstrap submitted the publication again on every restart, even when Rust +already held its durable outcome. Rust's existing idempotency made that replay +safe, but recovery unnecessarily depended on another successful write response. + +Go now observes the original publication before submitting it. This changes +activation recovery, not the authorization or provider execution protocol. +No dependency pin, governance DSL or Rust ledger format changes are required. + +* Authority and language boundaries +Scheme retains Agentic AI Lab governor and Governance semantics. Its existing +source-assets and policy bridge produce the admitted Cedar publication and +typed task/profile contracts. JSON and YAML use the same frozen Go bundle. +Go validates that bundle, registers its immutable tasks, then calls +=ObserveOrPublish= with its original publication identity. Task registration +does not activate ingress and does not authorize an external provider call. + +Rust remains the deciding owner of publication records, Policy/fence state, +claim consumption, shared budget and final provider dispatch/release checks. +Go does not synthesize a replacement parent, new operation ID or fresh claim +when recovery fails. Existing Rust publication replay semantics are reused. + +| Observation | Go action | Activation | +|-------------+-----------+------------| +| Exact committed ID, digest, revision and fence | Return original receipt; no PublishAuthority | Only after frozen bundle/view checks | +| Exact rejected decision | Return original rejection; no submission | Denied | +| Authoritative NotFound | Submit identical command once through Rust CAS | Only after validated committed response | +| Unavailable, denied, unsupported or other read failure | Return error; no submission | Denied | +| Malformed or mismatched receipt | Return binding error; no submission | Denied | +| Submission response lost | Return error; no automatic retry in this startup | Denied until another exact observation | + +NotFound is an observation, not an abort certificate. An earlier command may +still commit after that read. The subsequent identical command must therefore +reach the existing Rust idempotency/CAS boundary; the read and write are not +claimed to be one cross-language transaction. Concurrent identical startups +may each submit after NotFound. Rust must serialize them and preserve one +original outcome. Conflicting payloads under the same ID remain rejected. + +The parent value is copied before inspection so caller mutation cannot rebase +the submission. Production bootstrap already freezes its complete validated +bundle. The authenticated source endpoint and pinned method role remain trust +premises; a forged NotFound response from an untrusted server is not accepted +as an authority design. + +* Historical outcomes are not execution permission +A committed original receipt may remain readable after a newer revocation. +Recovering it only reestablishes Go's original routing view. Rust must reject +that stale view at fresh admission, consumption, dispatch and stream release +boundaries. Recovery does not reset a consumed logical claim, refund an unknown +reservation or redispatch a non-idempotent provider request. This change does +not add automatic policy hot reload or prove exactly-once remote execution. + +* Executable assurance +The gatewayclient unit tests cover committed and rejected observations, read +failures, malformed receipts, missing outcomes, lost submission responses and +parent mutation. The actual mTLS recovery test keeps a relay that drops every +successful publication response throughout recovery. Activation after restart +therefore demonstrates observation-based recovery rather than a lucky replay. + +The integration sequence restarts both owners, retries a consumed original +node and compares its canonical receipt without a second provider invocation. +It then publishes a later revocation and tests a different, unused original +node, distinguishing a fresh authority rejection from a spent-claim rejection. +The Scheme source-assets path runs the same recovery helper as the JSON path. + +| Asset | Responsibility | +|-------+----------------| +| =dev/proofs/quint/AuthorityPublication.qnt= | Bounded interleavings of observation, late commit, response loss, restart, activation, revocation and spent execution; explicit unsafe activation/source substitution/retry counterexamples | +| =dev/proofs/lean/PublicationRecovery.lean= | Inductive preservation of original decisions and accounting; exact payload/parent/fence observation survives the existing recovery relation | +| =dev/sysml/WeaverExecution.sysml= | Allocate observation-based recovery and execution freshness to their existing Go/Rust responsibilities | +| =59.03-provider-rpc-core.d2= | Show outcome lookup and initial CAS on the source lane, distinct from operation admission | + +The Quint submission counter checks Go's per-bootstrap operational contract. +Its known-outcome replay counterexample intentionally leaves Rust ledger and +execution safety intact: redundant recovery writes violate that narrower +contract, not Rust idempotency. Missing and failed reads cannot activate ingress. +Lean abstracts authenticated persistence and transport validation; actual +digest binding, durable restart behavior and fresh Cedar checks require the +native integration evidence. Neither model substitutes for those checks. + +Local validation is the acceptance path for this change. Remote GitHub CI is +not a blocker under the current private-project workflow. Exact commands, +source identity and results belong in the local qualification record; raw +logs and generated runtime products remain temporary or ignored. diff --git a/go/dev/provider_control_test.go b/go/dev/provider_control_test.go index b6f184de..c4e80fae 100644 --- a/go/dev/provider_control_test.go +++ b/go/dev/provider_control_test.go @@ -30,6 +30,7 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" "gopkg.in/yaml.v3" ) @@ -337,7 +338,8 @@ func TestProviderControlCompleteStreamsShareBudgetAndRejectSDKRetry(t *testing.T // Reuse the native publication reply-loss relay and its per-call counter. func (p *publicationReplyLoss) RegisterTask(ctx context.Context, in *pb.RegisterTaskRequest) (*pb.Empty, error) { - return p.source.RPC.RegisterTask(p.source.Context(ctx), in) + source := p.sourceClient() + return source.RPC.RegisterTask(source.Context(ctx), in) } func TestProviderControlPublicationReplyLossCannotActivateOrRebase(t *testing.T) { exerciseControlPublicationReplyLoss(t, nil) @@ -388,9 +390,9 @@ func exerciseControlPublicationReplyLoss(t *testing.T, source func(*testing.T, * exited := make(chan error, 1) go func() { exited <- server.Serve(listener) }() t.Cleanup(func() { server.Stop(); <-exited }) + c.env = append(c.env, "WEAVER_CONTROL_AUTHORITY_SOURCE_ADDRESS="+listener.Addr().String()) child := exec.Command(controlBinary) child.Env = append(os.Environ(), c.env...) - child.Env = append(child.Env, "WEAVER_CONTROL_AUTHORITY_SOURCE_ADDRESS="+listener.Addr().String()) log, err := os.Create(filepath.Join(c.directory, "lost-publication.log")) if err != nil { t.Fatal(err) @@ -413,7 +415,7 @@ func exerciseControlPublicationReplyLoss(t *testing.T, source func(*testing.T, * t.Error("failed bootstrap did not exit naturally") } log.Close() - if proxy.calls.Load() != 1 { + if proxy.calls.Load() != 1 || proxy.inspections.Load() != 1 { t.Fatal("bootstrap automatically retried publication") } ctx, cancel := context.WithTimeout(context.Background(), 12*time.Second) @@ -432,6 +434,7 @@ func exerciseControlPublicationReplyLoss(t *testing.T, source func(*testing.T, * cancel() f.stop(t) f.start(t) + proxy.replaceSource(f.admin) c.start(t) ctx, cancel = context.WithTimeout(context.Background(), 12*time.Second) defer cancel() @@ -441,4 +444,50 @@ func exerciseControlPublicationReplyLoss(t *testing.T, source func(*testing.T, * if f.primary.Load() != 1 { t.Fatal("recovery duplicated provider send") } + if proxy.calls.Load() != 1 || proxy.inspections.Load() != 2 { + t.Fatal("committed recovery submitted another publication") + } + key := &pb.OperationKey{Family: bundle.Plans[0].Family, Node: bundle.Plans[0].Node} + before, err := c.client.RPC.Inspect(c.client.Context(ctx), key) + if err != nil { + t.Fatal(err) + } + c.stop(t) + f.stop(t) + f.start(t) + proxy.replaceSource(f.admin) + c.start(t) + cancel() + ctx, cancel = context.WithTimeout(context.Background(), 12*time.Second) + defer cancel() + if _, err := c.client.Execute(ctx, bundle.Plans[0], pb.Encoding_PROTOBUF, 1, func(*pb.Event) error { return nil }); err == nil { + t.Fatal("dual restart reopened the consumed claim") + } + after, err := c.client.RPC.Inspect(c.client.Context(ctx), key) + if err != nil || !proto.Equal(before, after) || f.primary.Load() != 1 || proxy.calls.Load() != 1 || proxy.inspections.Load() != 3 { + t.Fatal("observed recovery changed original accounting or submitted publication", err) + } + c.stop(t) + parent := bundle.View().Revision + revocation := bundle.Publication + revocation.OperationID = "post-recovery-revocation" + revocation.ExpectedRevision = &parent + revocation.Fence++ + revocation.Policy = "forbid(principal,action,resource);" + if _, err := f.admin.Publish(ctx, revocation); err != nil { + t.Fatal(err) + } + f.stop(t) + f.start(t) + proxy.replaceSource(f.admin) + c.start(t) + cancel() + ctx, cancel = context.WithTimeout(context.Background(), 12*time.Second) + defer cancel() + if _, err := c.client.Execute(ctx, bundle.Plans[1], pb.Encoding_PROTOBUF, 1, func(*pb.Event) error { return nil }); err == nil { + t.Fatal("historical receipt overrode later revocation") + } + if f.primary.Load() != 1 || f.backup.Load() != 0 || proxy.calls.Load() != 1 || proxy.inspections.Load() != 4 { + t.Fatal("historical recovery sent or republished") + } } diff --git a/go/dev/publication_recovery_test.go b/go/dev/publication_recovery_test.go index 91aea565..363a75b1 100644 --- a/go/dev/publication_recovery_test.go +++ b/go/dev/publication_recovery_test.go @@ -7,6 +7,7 @@ import ( "crypto/tls" "errors" "net" + "sync" "sync/atomic" "testing" "time" @@ -24,13 +25,33 @@ import ( // Test-only mTLS relay loses the successful source reply after forwarding one RPC. type publicationReplyLoss struct { pb.UnimplementedGatewayServer - source *gatewayclient.Client - calls atomic.Int64 + source *gatewayclient.Client + sourceMu sync.RWMutex + calls atomic.Int64 + inspections atomic.Int64 +} + +func (s *publicationReplyLoss) sourceClient() *gatewayclient.Client { + s.sourceMu.RLock() + defer s.sourceMu.RUnlock() + return s.source +} +func (s *publicationReplyLoss) replaceSource(source *gatewayclient.Client) { + s.sourceMu.Lock() + defer s.sourceMu.Unlock() + s.source = source +} + +func (s *publicationReplyLoss) InspectPublication(ctx context.Context, key *pb.PublicationKey) (*pb.PublicationReceipt, error) { + s.inspections.Add(1) + source := s.sourceClient() + return source.RPC.InspectPublication(source.Context(ctx), key) } func (s *publicationReplyLoss) PublishAuthority(ctx context.Context, p *pb.PublishRequest) (*pb.PublicationReceipt, error) { s.calls.Add(1) - if _, err := s.source.RPC.PublishAuthority(s.source.Context(ctx), p); err != nil { + source := s.sourceClient() + if _, err := source.RPC.PublishAuthority(source.Context(ctx), p); err != nil { return nil, err } return nil, status.Error(codes.Unavailable, "injected reply loss after durable decision") diff --git a/go/internal/controlplane/gateway.go b/go/internal/controlplane/gateway.go index 4a13def8..0f7ec2e6 100644 --- a/go/internal/controlplane/gateway.go +++ b/go/internal/controlplane/gateway.go @@ -43,7 +43,7 @@ func Bootstrap(ctx context.Context, source, owner *gatewayclient.Client, bundle return nil, err } } - receipt, err := source.Publish(ctx, frozen.Publication) + receipt, err := source.ObserveOrPublish(ctx, frozen.Publication) if err != nil { return nil, err } diff --git a/go/internal/gatewayclient/publication.go b/go/internal/gatewayclient/publication.go index a9aed0da..cdd2cf5d 100644 --- a/go/internal/gatewayclient/publication.go +++ b/go/internal/gatewayclient/publication.go @@ -7,10 +7,33 @@ import ( pb "github.com/HardenedLinux/weaver/go/api/gateway/v1" v1 "github.com/HardenedLinux/weaver/go/api/v1" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) var ErrPublicationRejected = errors.New("original publication CAS rejected") +// ObserveOrPublish reads the original durable decision before one initial CAS. +// NotFound does not abort a possibly late command: the identical ID, payload and +// parent still go through Rust's final CAS. No failed Publish is retried here. +func (c *Client) ObserveOrPublish(ctx context.Context, p v1.Publication) (*pb.PublicationReceipt, error) { + if p.ExpectedRevision != nil { + parent := *p.ExpectedRevision + p.ExpectedRevision = &parent + } + receipt, err := c.InspectPublication(ctx, p) + if err == nil { + if receipt.Outcome == pb.PublicationOutcome_REJECTED { + return receipt, ErrPublicationRejected + } + return receipt, nil + } + if status.Code(err) != codes.NotFound { + return nil, err + } + return c.Publish(ctx, p) +} + func Publication(p v1.Publication) *pb.PublishRequest { return &pb.PublishRequest{OperationId: p.OperationID, ExpectedRevision: p.ExpectedRevision, Fence: p.Fence, Policy: p.Policy, Schema: p.Schema} } diff --git a/go/internal/gatewayclient/publication_activation_test.go b/go/internal/gatewayclient/publication_activation_test.go new file mode 100644 index 00000000..a578fcc4 --- /dev/null +++ b/go/internal/gatewayclient/publication_activation_test.go @@ -0,0 +1,115 @@ +package gatewayclient + +import ( + "context" + "errors" + "testing" + + pb "github.com/HardenedLinux/weaver/go/api/gateway/v1" + v1 "github.com/HardenedLinux/weaver/go/api/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type activationRPC struct { + pb.GatewayClient + inspect func(*pb.PublicationKey) (*pb.PublicationReceipt, error) + publish func(*pb.PublishRequest) (*pb.PublicationReceipt, error) + reads, writes int +} + +func (r *activationRPC) InspectPublication(_ context.Context, key *pb.PublicationKey, _ ...grpc.CallOption) (*pb.PublicationReceipt, error) { + r.reads++ + return r.inspect(key) +} +func (r *activationRPC) PublishAuthority(_ context.Context, p *pb.PublishRequest, _ ...grpc.CallOption) (*pb.PublicationReceipt, error) { + r.writes++ + return r.publish(p) +} + +func TestActivationReadsOriginalDecisionBeforeSubmission(t *testing.T) { + parent, fence, next := uint64(4), uint64(8), uint64(5) + proposal := v1.Publication{OperationID: "source-activation", ExpectedRevision: &parent, Fence: fence, Policy: "permit(principal,action,resource);", Schema: "{}"} + digest, err := proposal.Digest() + if err != nil { + t.Fatal(err) + } + good := func() *pb.PublicationReceipt { + return &pb.PublicationReceipt{OperationId: proposal.OperationID, Digest: digest, Outcome: pb.PublicationOutcome_COMMITTED, Revision: &next, Fence: &fence} + } + for _, outcome := range []pb.PublicationOutcome{pb.PublicationOutcome_COMMITTED, pb.PublicationOutcome_REJECTED} { + t.Run(outcome.String(), func(t *testing.T) { + rpc := &activationRPC{inspect: func(key *pb.PublicationKey) (*pb.PublicationReceipt, error) { + if key.OperationId != proposal.OperationID || key.Digest != digest { + t.Fatal("original query identity changed") + } + r := good() + r.Outcome = outcome + return r, nil + }} + r, err := (&Client{RPC: rpc}).ObserveOrPublish(t.Context(), proposal) + if rpc.reads != 1 || rpc.writes != 0 || r == nil { + t.Fatal("existing decision caused a source write") + } + if outcome == pb.PublicationOutcome_REJECTED && !errors.Is(err, ErrPublicationRejected) { + t.Fatal("rejected original decision activated", err) + } + if outcome == pb.PublicationOutcome_COMMITTED && err != nil { + t.Fatal(err) + } + }) + } + t.Run("absent outcome and lost submission reply", func(t *testing.T) { + rpc := &activationRPC{inspect: func(*pb.PublicationKey) (*pb.PublicationReceipt, error) { + return nil, status.Error(codes.NotFound, "not observed") + }, publish: func(p *pb.PublishRequest) (*pb.PublicationReceipt, error) { + if p.OperationId != proposal.OperationID || p.GetExpectedRevision() != parent || p.Fence != fence { + t.Fatal("submission rebased original command") + } + return nil, status.Error(codes.Unavailable, "unknown commit") + }} + if _, err := (&Client{RPC: rpc}).ObserveOrPublish(t.Context(), proposal); status.Code(err) != codes.Unavailable { + t.Fatal(err) + } + if rpc.reads != 1 || rpc.writes != 1 { + t.Fatal("ambiguous publication was retried") + } + }) + for _, code := range []codes.Code{codes.Unavailable, codes.PermissionDenied, codes.ResourceExhausted, codes.Unimplemented} { + t.Run(code.String(), func(t *testing.T) { + rpc := &activationRPC{inspect: func(*pb.PublicationKey) (*pb.PublicationReceipt, error) { + return nil, status.Error(code, "inspection failed") + }} + if _, err := (&Client{RPC: rpc}).ObserveOrPublish(t.Context(), proposal); status.Code(err) != code || rpc.writes != 0 { + t.Fatal("inspection failure caused publication", err) + } + }) + } + t.Run("mismatched observation", func(t *testing.T) { + rpc := &activationRPC{inspect: func(*pb.PublicationKey) (*pb.PublicationReceipt, error) { + r := good() + r.Digest = "substituted" + return r, nil + }} + if _, err := (&Client{RPC: rpc}).ObserveOrPublish(t.Context(), proposal); err == nil || rpc.writes != 0 { + t.Fatal("malformed observation activated or published") + } + }) + t.Run("frozen parent across inspection", func(t *testing.T) { + original := parent + rpc := &activationRPC{inspect: func(*pb.PublicationKey) (*pb.PublicationReceipt, error) { + parent++ + return nil, status.Error(codes.NotFound, "not observed") + }, publish: func(p *pb.PublishRequest) (*pb.PublicationReceipt, error) { + if p.GetExpectedRevision() != original { + t.Fatal("caller mutation replaced frozen parent") + } + return good(), nil + }} + defer func() { parent = original }() + if _, err := (&Client{RPC: rpc}).ObserveOrPublish(t.Context(), proposal); err != nil || rpc.writes != 1 { + t.Fatal(err) + } + }) +}