Async `Create` can destroy and re-create a live external resource: `Connect()` refills the shared TF state while an async create is in flight

### What happened?

For Terraform Plugin SDKv2 resources reconciled with the **async** external client, an in-flight async `Create` can **delete and re-create a live external resource** instead of creating a new one. No error is returned.

**The entry condition is a transient misobservation.** crossplane-runtime calls `Create` whenever `Observe` reports that the external resource does not exist. A one-off "not found" — a 404 blip, an eventually-consistent read — is what the Terraform provider's read maps to `d.SetId("")`, which makes upjet nil its cached Terraform state ([`external_tfpluginsdk.go#L508`](https://github.com/crossplane/upjet/blob/dc2f18a72484/pkg/controller/external_tfpluginsdk.go#L508)) and report `ResourceExists: false`. Crossplane then calls `Create` for an external resource that in fact still exists. On its own that is harmless — the create fails with "already exists" and the next reconcile self-heals. The bug is what happens when another reconcile interleaves with that create.

Two independent facts combine:

1. **The async `Create` goroutine reads the shared Terraform state at execution time, not at launch time.** [`external_async_tfpluginsdk.go#L180`](https://github.com/crossplane/upjet/blob/dc2f18a72484/pkg/controller/external_async_tfpluginsdk.go#L180) calls the sync `Create`, which evaluates `n.opTracker.GetTfState()` as an argument to `Apply()` inside the goroutine ([`external_tfpluginsdk.go#L654`](https://github.com/crossplane/upjet/blob/dc2f18a72484/pkg/controller/external_tfpluginsdk.go#L654)).
2. **`Connect()` has no in-flight-operation guard.** [`external_tfpluginsdk.go#L271-L315`](https://github.com/crossplane/upjet/blob/dc2f18a72484/pkg/controller/external_tfpluginsdk.go#L271-L315) reconstructs and writes the tracker state whenever `!opTracker.HasState()`, unconditionally. The `LastOperation.IsRunning()` guard exists only in `Observe()` ([`external_async_tfpluginsdk.go#L119`](https://github.com/crossplane/upjet/blob/dc2f18a72484/pkg/controller/external_async_tfpluginsdk.go#L119)) — and `Connect()` runs *before* `Observe()`, so that guard never covers this window.

So a reconcile that lands while an async create is queued refills the shared state with the external resource's real ID. When the goroutine finally runs, `Apply()` receives a state **with** an ID plus a **create diff** — and a create diff carries `RequiresNew` for any `ForceNew` argument that is either changing or newly computed, which in practice is every create diff for a resource that has `ForceNew` arguments at all. That is the terraform-plugin-sdk's "replace" contract ([`helper/schema/resource.go#L945`](https://github.com/hashicorp/terraform-plugin-sdk/blob/v2.38.2/helper/schema/resource.go#L945)):

```go
if d.Destroy || d.RequiresNew() {
if s.ID != "" {
diags = append(diags, r.delete(ctx, data, meta)...) // destroy the live resource
...
```

The two possible outcomes of that misobservation, depending only on who wins the window:

| State the goroutine reads | `Apply()` does | Outcome |
| --- | --- | --- |
| `nil` (goroutine wins the window) | plain create | harmless — usually a 409 `AlreadyExists`, then self-heals |
| state **with** ID (a `Connect()` slipped in) | **delete, then create** | external resource destroyed and replaced, silently |

For context on why nothing catches this today: `assertNoForceNew()` ([def L731](https://github.com/crossplane/upjet/blob/dc2f18a72484/pkg/controller/external_tfpluginsdk.go#L731), [call L764](https://github.com/crossplane/upjet/blob/dc2f18a72484/pkg/controller/external_tfpluginsdk.go#L764)) refuses to replace a resource on the **Update** path. It cannot simply be mirrored onto Create: a create diff legitimately carries `RequiresNew` — the SDK sets it whenever a `ForceNew` argument is changing **or** newly computed (`finalizeDiff`: `d.RequiresNew = d.Old != d.New || d.NewComputed`) — so an unconditional assertion there would reject every create. The Create path therefore passes whatever the shared cache holds straight to `Apply()`, and the ID in that state is what silently selects create vs. replace.

**`go test -race` does not catch this.** `AsyncTracker` guards its state with a mutex, so every access is properly synchronised — this is a logical race, not a data race.

#### Preconditions

1. The resource uses the async TF Plugin SDK external client.
2. An `Observe()` maps a transient "not found" to `d.SetId("")` — any 404-ish blip or eventual-consistency read. upjet nils the tracker state and crossplane-runtime then calls `Create` on a resource that in fact still exists.
3. Another reconcile's `Connect()` lands between the goroutine's launch and its state read. crossplane-runtime returns `Requeue: true` after a create, and the annotation writes around `Create` generate watch events, so a follow-up reconcile arrives within about a second.
4. The resource has at least one `ForceNew` argument. Since the SDK sets `RequiresNew` when such an argument is changing **or** newly computed, a create diff computed against an empty state carries it in practice always. Nearly all resources qualify.
5. The Terraform ID is derived deterministically from the spec (`NameAsIdentifier`, `ParameterAsIdentifier`, …), so the reconstructed state has a non-empty ID.

On (5) — a populated `status.atProvider` is **not** required. `Connect()` sets `tfState["id"] = params["id"]` unconditionally, so the refilled state carries a non-empty ID even when the observation is empty. We verified that variant separately; it is equally destructive.

#### Impact

Normally the goroutine reaches `Apply()` within microseconds, so the window is tiny and the harmless row of the table above is what you see. Under CPU pressure — for us, shortly after a provider pod restart — the window widened to tens of seconds and the destructive row was taken.

In our environment the benign variant fired 21 times over 5 months across the fleet, always self-healing via `AlreadyExists`, and therefore invisible. It then landed on the destructive path once, on an `aws_s3_bucket` MR with `force_destroy: true`: the delete purged every object version and succeeded, the bucket was re-created empty seconds later. The MR itself was healthy and untouched — nothing was being deleted, and no human action was involved.

The `force_destroy`-style flag is what turns this from a failed delete into data loss, but the replace itself is generic: any resource whose delete succeeds can be silently destroyed and re-created.

### How can we reproduce it?

The test below drives the real async external client and the real `terraform-plugin-sdk` `Apply()`. Only the resource's CRUD funcs and the provider setup are stubbed, so it hits no cloud API and needs no credentials. It uses exported API only.

```console
# in a checkout of crossplane/upjet
mkdir -p pkg/controller/createrace
# save the file below as pkg/controller/createrace/repro_test.go
go test ./pkg/controller/createrace/ -v
```

It runs two subtests that differ **only** in whether a second `Connect()` happens while the create is in flight:

```console
=== RUN TestAsyncCreateReplacesLiveResource/NoInterleavedConnect
repro_test.go:57: TF ID in the shared AsyncTracker when Apply() ran: ""
repro_test.go:58: downstream Terraform CRUD calls: [Read(my-resource) Create(my-resource)]
--- PASS: TestAsyncCreateReplacesLiveResource/NoInterleavedConnect

=== RUN TestAsyncCreateReplacesLiveResource/InterleavedConnect
repro_test.go:55: reconcile #2: TF ID after Connect() = "my-resource" <-- refilled while the create is in flight
repro_test.go:55: reconcile #2: Observe() -> exists=true upToDate=true (no error surfaced)
repro_test.go:55: async Create reported no error
repro_test.go:57: TF ID in the shared AsyncTracker when Apply() ran: "my-resource"
repro_test.go:58: downstream Terraform CRUD calls: [Read(my-resource) Delete(my-resource, purge_on_delete=true) Create(my-resource)]
repro_test.go:62: the async Create destroyed the live external resource.
--- FAIL: TestAsyncCreateReplacesLiveResource/InterleavedConnect
```

The goroutine is parked deterministically by installing a `logging.Logger` on the `OperationTrackerStore` that blocks on the `"Async create starting..."` line — the statement immediately preceding the state read. That only removes the timing flakiness; nothing in upjet or controller-runtime guarantees the goroutine wins this window in production.

pkg/controller/createrace/repro_test.go

```go
// SPDX-FileCopyrightText: 2026 The Crossplane Authors
//
// SPDX-License-Identifier: Apache-2.0

// Package createrace reproduces a logical race between
// TerraformPluginSDKConnector.Connect() and an in-flight async Create
// goroutine, which can make the async Create destroy and re-create a live
// external resource instead of creating a new one.
//
// Run with:
//
// go test ./pkg/controller/createrace/ -v
//
// Note: -race reports nothing here. AsyncTracker guards its state with a
// mutex, so this is a logical race, not a data race.
package createrace

import (
"context"
"fmt"
"strings"
"sync"
"sync/atomic"
"testing"
"time"

"github.com/crossplane/crossplane-runtime/v2/pkg/logging"
"github.com/crossplane/crossplane-runtime/v2/pkg/meta"
xpresource "github.com/crossplane/crossplane-runtime/v2/pkg/resource"
"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"

"github.com/crossplane/upjet/v2/pkg/config"
tjcontroller "github.com/crossplane/upjet/v2/pkg/controller"
"github.com/crossplane/upjet/v2/pkg/resource/fake"
"github.com/crossplane/upjet/v2/pkg/terraform"
)

const externalName = "my-resource"

// TestAsyncCreateReplacesLiveResource asserts that an async Create only ever
// creates. It fails on the "interleaved Connect" case, where the async Create
// deletes the live external resource first.
func TestAsyncCreateReplacesLiveResource(t *testing.T) {
for _, tc := range []struct {
name string
interleaveConnect bool
}{
{name: "NoInterleavedConnect", interleaveConnect: false},
{name: "InterleavedConnect", interleaveConnect: true},
} {
t.Run(tc.name, func(t *testing.T) {
calls, tfID := reconcile(t, tc.interleaveConnect)

t.Logf("TF ID in the shared AsyncTracker when Apply() ran: %q", tfID)
t.Logf("downstream Terraform CRUD calls: %v", calls)

for _, c := range calls {
if strings.HasPrefix(c, "Delete") {
t.Errorf(`the async Create destroyed the live external resource.
got calls: %v
want calls: [Read Create]
Apply() was handed a state with ID %q plus a diff with RequiresNew, so the
TF plugin SDK took its "replace" path: delete-then-create.`, calls, tfID)
}
}
})
}
}

// reconcile drives two reconciles of the same MR against the real async
// external client.
//
// Reconcile #1: the external resource is misobserved as gone (a 404 from an
// existence check that the resource's Read maps to d.SetId("")), so the
// managed reconciler would call Create. upjet launches its async Create
// goroutine and returns. The goroutine is parked before it reads the shared
// state, modelling scheduling delay under CPU pressure.
//
// Reconcile #2 (when interleaveConnect is set): Connect() finds the tracker's
// state empty and reconstructs it from status.atProvider, writing an ID back
// into the shared tracker. Observe() then short-circuits on
// LastOperation.IsRunning() and reports the resource as existing and
// up-to-date, so no error surfaces.
//
// The goroutine is then released and reads whatever state the tracker holds.
func reconcile(t *testing.T, interleaveConnect bool) ([]string, string) {
t.Helper()

rec := &recorder{}
notFound := &atomic.Bool{}

cfg := &config.Resource{
Name: "example_resource",
TerraformResource: exampleResource(rec, notFound),
// Any deterministic, spec-derived ID reproduces this; e.g.
// aws_s3_bucket uses config.ParameterAsIdentifier("bucket").
ExternalName: config.NameAsIdentifier,
Sensitive: config.Sensitive{AdditionalConnectionDetailsFn: func(_ map[string]any) (map[string][]byte, error) {
return nil, nil
}},
}

mg := newMR()
g := newGate()
cb := newCallbacks()

// The gate is installed as the OperationTrackerStore's logger: the async
// Create goroutine logs "Async create starting..." through it immediately
// before calling the sync Create, which is where the shared state is read.
ots := tjcontroller.NewOperationStore(g)
conn := tjcontroller.NewTerraformPluginSDKAsyncConnector(nil, ots, setup, cfg,
tjcontroller.WithTerraformPluginSDKAsyncLogger(logging.NewNopLogger()),
tjcontroller.WithTerraformPluginSDKAsyncCallbackProvider(cb),
)

ctx := context.Background()

// ---- reconcile #1 ----------------------------------------------------
ec1, err := conn.Connect(ctx, mg)
if err != nil {
t.Fatalf("reconcile #1 Connect(): %v", err)
}
t.Logf("reconcile #1: TF ID after Connect() = %q", ots.Tracker(mg).GetTfID())

notFound.Store(true) // the misobservation
obs, err := ec1.Observe(ctx, mg)
if err != nil {
t.Fatalf("reconcile #1 Observe(): %v", err)
}
if obs.ResourceExists {
t.Fatal("reconcile #1: expected the misobservation to report the resource as absent")
}
t.Logf("reconcile #1: Observe() -> exists=false; TF ID now = %q", ots.Tracker(mg).GetTfID())

// The resource was never actually gone: the 404 was a one-off.
notFound.Store(false)

if _, err := ec1.Create(ctx, mg); err != nil {
t.Fatalf("reconcile #1 Create(): %v", err)
}
waitFor(t, "the async Create goroutine to park before Apply()", g.reached)

// ---- reconcile #2 ----------------------------------------------------
if interleaveConnect {
ec2, err := conn.Connect(ctx, mg)
if err != nil {
t.Fatalf("reconcile #2 Connect(): %v", err)
}
t.Logf("reconcile #2: TF ID after Connect() = %q <-- refilled while the create is in flight", ots.Tracker(mg).GetTfID())

obs2, err := ec2.Observe(ctx, mg)
if err != nil {
t.Fatalf("reconcile #2 Observe(): %v", err)
}
t.Logf("reconcile #2: Observe() -> exists=%v upToDate=%v (no error surfaced)", obs2.ResourceExists, obs2.ResourceUpToDate)
}

tfID := ots.Tracker(mg).GetTfID()

// ---- release the goroutine -------------------------------------------
g.disarm()
close(g.release)
waitFor(t, "the async Create to complete", cb.done)
if err := cb.err.Load(); err != nil {
t.Logf("async Create reported: %v", err)
} else {
t.Log("async Create reported no error")
}

return rec.calls(), tfID
}

// exampleResource is a minimal SDKv2 resource with a ForceNew identifier
// argument. Its CRUD funcs only record which downstream call the SDK made.
func exampleResource(rec *recorder, notFound *atomic.Bool) *schema.Resource {
return &schema.Resource{
Schema: map[string]*schema.Schema{
"name": {
Type: schema.TypeString,
Optional: true,
Computed: true,
ForceNew: true,
},
// Stands in for aws_s3_bucket's force_destroy: it makes the
// delete succeed instead of being refused.
"purge_on_delete": {
Type: schema.TypeBool,
Optional: true,
},
},
CreateWithoutTimeout: func(_ context.Context, d *schema.ResourceData, _ any) diag.Diagnostics {
n, _ := d.Get("name").(string)
rec.add("Create(" + n + ")")
d.SetId(n)
return nil
},
ReadWithoutTimeout: func(_ context.Context, d *schema.ResourceData, _ any) diag.Diagnostics {
rec.add("Read(" + d.Id() + ")")
if notFound.Load() {
d.SetId("") // "not found, removing from state"
}
return nil
},
UpdateWithoutTimeout: func(_ context.Context, d *schema.ResourceData, _ any) diag.Diagnostics {
rec.add("Update(" + d.Id() + ")")
return nil
},
DeleteWithoutTimeout: func(_ context.Context, d *schema.ResourceData, _ any) diag.Diagnostics {
rec.add(fmt.Sprintf("Delete(%s, purge_on_delete=%v)", d.Id(), d.Get("purge_on_delete")))
d.SetId("")
return nil
},
}
}

// newMR returns an MR that has been Ready for a long time: its spec holds the
// deterministic name and status.atProvider holds the last good observation.
func newMR() *fake.Terraformed {
mg := &fake.Terraformed{
Parameterizable: fake.Parameterizable{Parameters: map[string]any{
"name": externalName,
"purge_on_delete": true,
}},
Observable: fake.Observable{Observation: map[string]any{
"id": externalName,
"name": externalName,
}},
}
mg.SetName("example")
mg.SetUID(types.UID("6f1b0c48-1f2a-4c1e-9f1a-000000000001"))
meta.SetExternalName(mg, externalName)
return mg
}

func setup(_ context.Context, _ client.Client, _ xpresource.Managed) (terraform.Setup, error) {
return terraform.Setup{}, nil
}

// gate is a logging.Logger that parks the async Create goroutine at the
// "Async create starting..." log line, i.e. after MarkStart("create") and
// before the shared state is read and handed to Apply(). It only makes the
// interleaving deterministic; nothing in upjet or controller-runtime
// guarantees the goroutine wins this window in production.
type gate struct {
reached chan struct{}
release chan struct{}
once sync.Once
armed atomic.Bool
}

func newGate() *gate {
g := &gate{reached: make(chan struct{}), release: make(chan struct{})}
g.armed.Store(true)
return g
}

func (g *gate) disarm() { g.armed.Store(false) }

func (g *gate) Debug(msg string, _ ...any) {
if g.armed.Load() && strings.Contains(msg, "Async create starting") {
g.once.Do(func() { close(g.reached) })
<-g.release
}
}

func (g *gate) Info(_ string, _ ...any) {}
func (g *gate) WithValues(_ ...any) logging.Logger { return g }

type callbacks struct {
done chan struct{}
err atomic.Value
}

func newCallbacks() *callbacks { return &callbacks{done: make(chan struct{})} }

func (c *callbacks) fn() terraform.CallbackFn {
return func(err error, _ context.Context) error {
if err != nil {
c.err.Store(err.Error())
}
close(c.done)
return nil
}
}

func (c *callbacks) Create(_ types.NamespacedName, _ bool) terraform.CallbackFn { return c.fn() }
func (c *callbacks) Update(_ types.NamespacedName, _ bool) terraform.CallbackFn { return c.fn() }
func (c *callbacks) Destroy(_ types.NamespacedName, _ bool) terraform.CallbackFn { return c.fn() }

type recorder struct {
mu sync.Mutex
c []string
}

func (r *recorder) add(s string) {
r.mu.Lock()
defer r.mu.Unlock()
r.c = append(r.c, s)
}

func (r *recorder) calls() []string {
r.mu.Lock()
defer r.mu.Unlock()
out := make([]string, len(r.c))
copy(out, r.c)
return out
}

func waitFor(t *testing.T, what string, ch <-chan struct{}) {
t.Helper()
select {
case <-ch:
case <-time.After(30 * time.Second):
t.Fatalf("timed out waiting for %s", what)
}
}
```

### Possible directions

Deferring to maintainers on the right shape. The two that seem sound to us:

1. **Snapshot the state at launch time.** Read `opTracker.GetTfState()` in the async `Create`/`Update` before starting the goroutine and pass it in, so the operation applies the diff against the state the diff was computed from. This addresses the root cause directly — a diff computed against state A is currently applied to state B — and it holds regardless of what else may write to the tracker.
2. **Guard `Connect()` with `LastOperation.IsRunning()`**, skipping the state reconstruction while an async operation is in flight, symmetric with the existing `Observe()` guard. This closes the specific window; (1) is the more robust of the two because it does not depend on enumerating the writers.

### Environment

- upjet `v2.2.1-0.20260610110527-59c45527ebe4` (where we hit it) and reproduced on `main` @ `dc2f18a72484`
- `github.com/hashicorp/terraform-plugin-sdk/v2` v2.38.2
- Go 1.26.4
- Observed with `provider-upjet-aws` (`aws_s3_bucket`), but the mechanism is provider-agnostic

Start by running the supplied reproduction in pkg/controller/createrace/repro_test.go, then read the async create path in pkg/controller/external_async_tfpluginsdk.go and Connect/Create in pkg/controller/external_tfpluginsdk.go. Trace how AsyncTracker state reaches Apply() and use the interleaved-Connect case as the regression target. Done means the async create never deletes the live resource and the reproduction passes without requiring cloud credentials.

Written by the indexing model from the issue text.