diff --git a/.agents/skills/loom-framework/SKILL.md b/.agents/skills/loom-framework/SKILL.md index 7783adc27..c4f99fc43 100644 --- a/.agents/skills/loom-framework/SKILL.md +++ b/.agents/skills/loom-framework/SKILL.md @@ -75,6 +75,14 @@ consumer validation. ## Architecture Boundaries +- `execution/` owns typed activity names, options, registration/execution + contracts, and futures. `execution/adapter` owns concrete-handler reflection; + it does not own serialization. `execution/selvedge` uses native generic + definitions and captures each future's workflow scope. Selvedge owns codecs, + retry execution, storage, and app lifecycle. Agent policies and type-erased + agent JSON bridges remain in their consumer. This runtime surface has no DSL + or generated workflow API. Keep the binding model and scope tests aligned. + - `internal/patternsyntax` owns the portable Go/ECMAScript bridge. Export derives syntax from the admitted Go tree; import admits a bounded ECMAScript grammar before Go parsing and translates direction-specific escapes. Never use Go diff --git a/.agents/skills/loom-framework/references/formal-models.md b/.agents/skills/loom-framework/references/formal-models.md index 9a87c3d8d..9a756c4a7 100644 --- a/.agents/skills/loom-framework/references/formal-models.md +++ b/.agents/skills/loom-framework/references/formal-models.md @@ -199,6 +199,7 @@ Rendered schema validation and actual decoder behavior are separate obligations. | Code concern | Model | | --- | --- | +| Activity registration publication and immutable future scopes | [Activity binding](../../../../execution/selvedge/tla/README.md); native-conflict and parent/canceled/detached scope tests, with a rejected future-rebinding alternative | | Artifact write intent, atomic create-only installation, rollback ownership and gen-tree admission | [Artifact write intent](../../../../cmd/loom/tla/artifact_write_intent/README.md); bounded intent-loss/race controls paired with real filesystem and registered-plugin CLI tests | | Service-error merge and history snapshot ownership | [Error ownership](../../../../pkg/tla/error_ownership/README.md); bounded aliasing counterexamples, with repeated merges and protobuf reconstruction checked in Go | | Public message ownership, history masking, diagnostic exclusion, transparent wrappers and independent joins | [Public disclosure](../../../../pkg/tla/error_disclosure/README.md); finite owner/message model, with HTTP/gRPC/JSON-RPC unary/SSE/WebSocket wire tests | diff --git a/.agents/skills/loom-framework/references/repo-map.md b/.agents/skills/loom-framework/references/repo-map.md index 515ea17cc..c9d32cc91 100644 --- a/.agents/skills/loom-framework/references/repo-map.md +++ b/.agents/skills/loom-framework/references/repo-map.md @@ -27,6 +27,8 @@ belong in the `loom` skill and the canonical guides under `docs/`. WebSocket lifecycle, including the client connection demultiplexer (`websocket_client.go`, modeled in `jsonrpc/tla/`); typed transport generation; and integrations +- `execution/`: typed activity contracts, adapter reflection, and native Selvedge + activity binding; no workflow DSL or persistence implementation - `observability/`: framework-owned tracing, metrics, logging, and transport event contracts - `security/authorization.go`: mandatory endpoint checks and cancellation; diff --git a/.agents/skills/loom/SKILL.md b/.agents/skills/loom/SKILL.md index b5e1fd872..4bbd96335 100644 --- a/.agents/skills/loom/SKILL.md +++ b/.agents/skills/loom/SKILL.md @@ -39,6 +39,20 @@ fields in the body, map them explicitly, or declare `ErrorNameFromStatus()` on an HTTP error response with a unique status. Views retain their projected contract. Generation rejects incomplete mappings; add the missing source and regenerate. +## Typed background activities + +Use `execution.Activity[Input, Output]` for named, typed activity calls without +adding DSL declarations. With Selvedge, create `execution/selvedge.New(app)`, +register each handler using its typed `Register` function and native options +before `app.Start`, then execute through `activities.Scope(workflowContext)`. +Native Selvedge owns codecs, workflow definitions, app lifecycle, submission, +status, cancellation, and transactions. Futures retain their scheduling scope. +The typed native registry does not implement `ActivityRegistrar`; that interface +serves backends able to register a concrete function dynamically. See +[typed activities](../../../docs/execution.md) for the complete boundary. +Selvedge v0.7.0 needs Go 1.27.3 and a fresh database; finish old runs and retain +their database before changing worker storage. + ## Core Workflow 1. Treat `design/*.go` as the source of truth. diff --git a/cmd/loom/gen_workspace_test.go b/cmd/loom/gen_workspace_test.go index 6a4985f6c..a62ecff27 100644 --- a/cmd/loom/gen_workspace_test.go +++ b/cmd/loom/gen_workspace_test.go @@ -19,9 +19,9 @@ func TestGeneratorUsesCallerWorkspace(t *testing.T) { root := t.TempDir() moduleDir := filepath.Join(root, "app") require.NoError(t, os.MkdirAll(filepath.Join(moduleDir, "design"), 0o755)) - mod := []byte("module example.com/service\n\ngo 1.27.2\n\nrequire github.com/CaliLuke/loom v0.0.0\n") + mod := []byte("module example.com/service\n\ngo 1.27.3\n\nrequire github.com/CaliLuke/loom v0.0.0\n") if workspace == "use" { - mod = []byte("module example.com/service\n\ngo 1.27.2\n") + mod = []byte("module example.com/service\n\ngo 1.27.3\n") } require.NoError(t, os.WriteFile(filepath.Join(moduleDir, "go.mod"), mod, 0o600)) sum, err := os.ReadFile(filepath.Join(repositoryRoot, "go.sum")) @@ -31,7 +31,7 @@ func TestGeneratorUsesCallerWorkspace(t *testing.T) { import . "github.com/CaliLuke/loom/dsl" var _ = API("service", func() {}) `), 0o600)) - work := "go 1.27.2\n\nuse ./app\n" + work := "go 1.27.3\n\nuse ./app\n" if workspace == "replace" { work += fmt.Sprintf("replace github.com/CaliLuke/loom => %q\n", repositoryRoot) } else { diff --git a/cmd/loom/generation_filesystem_test.go b/cmd/loom/generation_filesystem_test.go index e822a2eb6..9e7fffc9f 100644 --- a/cmd/loom/generation_filesystem_test.go +++ b/cmd/loom/generation_filesystem_test.go @@ -20,7 +20,7 @@ func TestGenerationStageUsesWritableOutput(t *testing.T) { parent := t.TempDir() module := filepath.Join(parent, "module") require.NoError(t, os.Mkdir(module, 0o750)) - require.NoError(t, os.WriteFile(filepath.Join(module, "go.mod"), []byte("module example.com/app\n\ngo 1.27.2\n"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(module, "go.mod"), []byte("module example.com/app\n\ngo 1.27.3\n"), 0o600)) require.NoError(t, os.Chmod(parent, 0o500)) t.Cleanup(func() { require.NoError(t, os.Chmod(parent, 0o700)) @@ -38,7 +38,7 @@ func TestGenerationStageModuleContext(t *testing.T) { output := filepath.Join(module, nested) require.NoError(t, os.MkdirAll(output, 0o750)) absoluteTarget := filepath.Join(root, "absolute") - original := []byte(fmt.Sprintf("module example.com/app\n\ngo 1.27.2\n\nreplace (\nexample.com/dep => ../dep\nexample.com/other v1.0.0 => ./other\nexample.com/absolute => %q\nexample.com/version v1.0.0 => example.com/release v1.2.3\n)\n", absoluteTarget)) + original := []byte(fmt.Sprintf("module example.com/app\n\ngo 1.27.3\n\nreplace (\nexample.com/dep => ../dep\nexample.com/other v1.0.0 => ./other\nexample.com/absolute => %q\nexample.com/version v1.0.0 => example.com/release v1.2.3\n)\n", absoluteTarget)) require.NoError(t, os.WriteFile(filepath.Join(module, "go.mod"), original, 0o600)) tx, err := newGenerationTransaction(output) require.NoError(t, err) @@ -91,14 +91,14 @@ func TestGenerationMountedDestinations(t *testing.T) { case "module": output = mount case "nested-output": - require.NoError(t, os.WriteFile(filepath.Join(base, "go.mod"), []byte("module example.com/app\n\ngo 1.27.2\n"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(base, "go.mod"), []byte("module example.com/app\n\ngo 1.27.3\n"), 0o600)) output = filepath.Join(base, "nested") require.NoError(t, os.Symlink(mount, output)) default: require.NoError(t, os.Symlink(mount, filepath.Join(output, "plugin"))) } if kind == "module" { - require.NoError(t, os.WriteFile(filepath.Join(output, "go.mod"), []byte("module example.com/app\n\ngo 1.27.2\n"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(output, "go.mod"), []byte("module example.com/app\n\ngo 1.27.3\n"), 0o600)) } t.Cleanup(func() { require.NoError(t, os.RemoveAll(filepath.Join(mount, "go.mod"))) diff --git a/cmd/loom/provenance_test.go b/cmd/loom/provenance_test.go index 4d366f3a7..21a486958 100644 --- a/cmd/loom/provenance_test.go +++ b/cmd/loom/provenance_test.go @@ -15,7 +15,7 @@ import ( // through the package loader before embedding the origin in generation or vet. func TestGeneratorCanonicalDesignOrigin(t *testing.T) { dir := t.TempDir() - require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte("module example.com/origin\n\ngo 1.27.2\n"), 0600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte("module example.com/origin\n\ngo 1.27.3\n"), 0600)) require.NoError(t, os.Mkdir(filepath.Join(dir, "design"), 0750)) require.NoError(t, os.WriteFile(filepath.Join(dir, "design", "design.go"), []byte("package design\n"), 0600)) t.Chdir(dir) diff --git a/codegen/generator/cli_argument_order_test.go b/codegen/generator/cli_argument_order_test.go index 63d06fe03..ae99b54bc 100644 --- a/codegen/generator/cli_argument_order_test.go +++ b/codegen/generator/cli_argument_order_test.go @@ -36,7 +36,7 @@ func TestGeneratedCLIArgumentOrder(t *testing.T) { require.NoError(t, os.MkdirAll(dir, 0o700)) } codegen.RunDSL(t, testdata.CLIArgumentOrderDSL(tc.transport, tc.shape)) - mod := fmt.Sprintf("module example.com/order\n\ngo 1.27.2\n\nrequire github.com/CaliLuke/loom v1.0.0\nreplace github.com/CaliLuke/loom => %s\n", loomModuleSource(t)) + mod := fmt.Sprintf("module example.com/order\n\ngo 1.27.3\n\nrequire github.com/CaliLuke/loom v1.0.0\nreplace github.com/CaliLuke/loom => %s\n", loomModuleSource(t)) require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte(mod), 0o600)) _, err := Generate(dir, "gen", "github.com/CaliLuke/loom/codegen/generator", false) require.NoError(t, err) diff --git a/codegen/generator/declaration_merge_artifacts_test.go b/codegen/generator/declaration_merge_artifacts_test.go index b8b9476c7..1a9f47d0e 100644 --- a/codegen/generator/declaration_merge_artifacts_test.go +++ b/codegen/generator/declaration_merge_artifacts_test.go @@ -63,7 +63,7 @@ type Moment time.Time dir := t.TempDir() _, err := mergeFilesByPath(files)[0].Render(dir) require.NoError(t, err) - require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte("module example.com/merge\n\ngo 1.27.2\n"), 0600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte("module example.com/merge\n\ngo 1.27.3\n"), 0600)) const test = `package shared import ("reflect"; "testing") func TestValues(t *testing.T) { diff --git a/codegen/generator/declaration_merge_specimens_test.go b/codegen/generator/declaration_merge_specimens_test.go index a4c671a6c..a13df5418 100644 --- a/codegen/generator/declaration_merge_specimens_test.go +++ b/codegen/generator/declaration_merge_specimens_test.go @@ -28,7 +28,7 @@ func TestSharedDeclarationSpecimens(t *testing.T) { dir = filepath.Join(root, shape) require.NoError(t, os.MkdirAll(dir, 0750)) } - module := fmt.Sprintf("module example.com/merge\n\ngo 1.27.2\n\nrequire github.com/CaliLuke/loom v0.0.0\nreplace github.com/CaliLuke/loom => %s\n", loomModuleSource(t)) + module := fmt.Sprintf("module example.com/merge\n\ngo 1.27.3\n\nrequire github.com/CaliLuke/loom v0.0.0\nreplace github.com/CaliLuke/loom => %s\n", loomModuleSource(t)) require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte(module), 0600)) _, err := Generate(dir, "gen", "github.com/CaliLuke/loom/codegen/generator", false) require.NoError(t, err) diff --git a/codegen/generator/example_cli_transport_dispatch_test.go b/codegen/generator/example_cli_transport_dispatch_test.go index b558fc52b..a202892d5 100644 --- a/codegen/generator/example_cli_transport_dispatch_test.go +++ b/codegen/generator/example_cli_transport_dispatch_test.go @@ -19,7 +19,7 @@ import ( func TestExampleCLISelectsTransportByCommand(t *testing.T) { codegen.RunDSL(t, cliTransportDispatchDesign) dir := t.TempDir() - goMod := fmt.Sprintf("module example.com/dispatch\n\ngo 1.27.2\n\nrequire github.com/CaliLuke/loom v0.0.0\n\nreplace github.com/CaliLuke/loom => %s\n", loomModuleSource(t)) + goMod := fmt.Sprintf("module example.com/dispatch\n\ngo 1.27.3\n\nrequire github.com/CaliLuke/loom v0.0.0\n\nreplace github.com/CaliLuke/loom => %s\n", loomModuleSource(t)) require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte(goMod), 0o600)) _, err := Generate(dir, "gen", "github.com/CaliLuke/loom/codegen/generator", false) @@ -89,7 +89,7 @@ func TestExampleCLISelectsTransportByCommand(t *testing.T) { func TestHTTPOnlyExampleCLICompiles(t *testing.T) { codegen.RunDSL(t, httpOnlyExampleCLIDesign) dir := t.TempDir() - goMod := fmt.Sprintf("module example.com/http-only\n\ngo 1.27.2\n\nrequire github.com/CaliLuke/loom v0.0.0\n\nreplace github.com/CaliLuke/loom => %s\n", loomModuleSource(t)) + goMod := fmt.Sprintf("module example.com/http-only\n\ngo 1.27.3\n\nrequire github.com/CaliLuke/loom v0.0.0\n\nreplace github.com/CaliLuke/loom => %s\n", loomModuleSource(t)) require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte(goMod), 0o600)) _, err := Generate(dir, "gen", "github.com/CaliLuke/loom/codegen/generator", false) diff --git a/codegen/generator/provenance_readiness_test.go b/codegen/generator/provenance_readiness_test.go index 0c464c20c..938bc659a 100644 --- a/codegen/generator/provenance_readiness_test.go +++ b/codegen/generator/provenance_readiness_test.go @@ -44,7 +44,7 @@ func TestGenerationProvenanceIncludesPluginState(t *testing.T) { }) require.NoError(t, eval.Register(&provenancePluginRoot{Value: value})) dir := t.TempDir() - require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte("module example.com/provenance\n\ngo 1.27.2\n"), 0600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte("module example.com/provenance\n\ngo 1.27.3\n"), 0600)) _, err := Generate(dir, "gen", "github.com/CaliLuke/loom/codegen/generator", false) require.NoError(t, err) data, err := os.ReadFile(filepath.Join(dir, codegen.Gendir, "loom.json")) diff --git a/docs/_index.md b/docs/_index.md index 9b5ff8953..7fcefa88c 100644 --- a/docs/_index.md +++ b/docs/_index.md @@ -100,6 +100,7 @@ func (s *helloService) SayHello(ctx context.Context, name string) (string, error | Guide | Description | |-------|-------------| +| [Typed activities](execution.md) | Typed execution contracts and native Selvedge activity binding | | [Quickstart](quickstart.md) | Install Loom and build your first service | | [DSL Reference](dsl-reference.md) | Complete reference for Loom's design language | | [Code Generation](code-generation.md) | Understanding Loom's code generation process | diff --git a/docs/execution.md b/docs/execution.md new file mode 100644 index 000000000..b84fd2542 --- /dev/null +++ b/docs/execution.md @@ -0,0 +1,72 @@ +# Typed activities + +`github.com/CaliLuke/loom/execution` provides a small typed activity contract for +workflow adapters. Use it when applications share typed activity definitions +and calls across execution backends. It is an ordinary Go runtime +package; it requires no DSL declarations or generated code. + +An activity couples a stable name with its input and output types: + +```go +readMetadata := execution.Activity[MetadataInput, MetadataOutput]{ + Name: "assets.read_metadata", +} +``` + +Register its handler through `readMetadata.Register(ctx, registrar, options, +handler)`. Inside a workflow, `readMetadata.Execute(ctx, executor, input, options)` +returns `MetadataOutput`; `ExecuteAsync` returns `Future[MetadataOutput]`. +Check every returned error. Inputs must be non-nil, and input and output types +must be supported by the selected backend's codec. Names must remain stable +across worker restarts and replay, and names are unique within a registry. + +`ActivityOptions` contains queue, retry, attempt, queue-wait, and heartbeat +settings. Zero fields leave values to registration or backend defaults; they do +not explicitly clear inherited settings. This contract does not make different +backends' retry or cancellation behavior identical. + +## Selvedge + +`github.com/CaliLuke/loom/execution/selvedge` adapts these activities to Selvedge +v0.7.0. Create an activity registry with `selvedge.New(app)` and register all +handlers before `app.Start`. The app remains caller-owned: the adapter never +starts or stops it. Register through the typed `selvedge.Register` function so +Selvedge retains the concrete input and output types and owns their codecs. +This registry does not implement the type-erased `ActivityRegistrar` interface. + +Inside a native Selvedge workflow, pass `activities.Scope(workflowContext)` as +the executor. Scope controls cancellation and detached execution. The standard +`context.Context` parameter does not replace that native workflow scope, and a +future always waits using its original scope. Activity JSON inputs and outputs +remain plain values without an extra envelope. + +The adapter's typed `selvedge.Register(activities, activity, handler, +nativeOptions)` function accepts the full native `selvedge.ActivityOptions`, +including maximum retry intervals and non-retryable error predicates. That path +preserves native defaults and error types. Agent-specific error conversion and +timeout categories belong in the consuming adapter. + +Keep workflows, start options, status inspection, cancellation requests, and +transaction-bound submission on the native Selvedge API. In particular, passing +the existing transaction context to native `StartWorkflow` keeps admission in +that transaction. The activity adapter neither starts another transaction nor +turns submission into a separate background call. + +Selvedge v0.7.0 requires Go 1.27.3 and a fresh database. Finish runs on the old +binary, retain its database for inspection, and configure a new database before +starting the upgraded workers. There is no migration from earlier development +schemas. + +## Adapter authors + +Implement `ActivityExecutor` and either a typed native registration function or +`ActivityRegistrar`, according to the backend API. Generic registration receives a +concrete handler and execution receives the concrete input plus a result pointer. +Preserve those types when the native serializer needs them. The +`execution/adapter` package provides the shared handler validation and binding +used by adapters; it is not an application scheduling API. + +This package intentionally has no general workflow definition, service-method +submission DSL, scheduler, persistence engine, or agent lifecycle. Applications +can use the common activity contract while retaining the native capabilities +their workflows need. diff --git a/docs/grpc-guide.md b/docs/grpc-guide.md index 46e71317f..768b2efd5 100644 --- a/docs/grpc-guide.md +++ b/docs/grpc-guide.md @@ -21,7 +21,7 @@ Loom's gRPC support includes: ### Prerequisites and Supported Toolchain -Use Go 1.27.2 and keep these protobuf tools on `PATH` when generating a +Use Go 1.27.3 and keep these protobuf tools on `PATH` when generating a gRPC service: | Tool | Supported version | diff --git a/execution/activity.go b/execution/activity.go new file mode 100644 index 000000000..208a04e8f --- /dev/null +++ b/execution/activity.go @@ -0,0 +1,94 @@ +// Package execution defines typed activity contracts for workflow adapters. +// It does not own persistence, workflow scheduling, or application policy. +package execution + +import "context" + +type ( + // ActivityRegistrar installs activities before workers begin polling. + ActivityRegistrar interface { + // RegisterActivity validates and installs a concrete handler. + RegisterActivity(context.Context, ActivityDefinition) error + } + + // ActivityExecutor schedules named activities in a workflow scope. + // Results must be decoded into the supplied pointer using the backend's codec. + ActivityExecutor interface { + // ExecuteActivity schedules and waits for a result, decoded into a pointer. + ExecuteActivity(context.Context, ActivityCall, any) error + // ExecuteActivityAsync schedules a command in this executor's workflow scope. + ExecuteActivityAsync(context.Context, ActivityCall) (ActivityFuture, error) + } + + // ActivityDefinition describes an activity independently of agent payload types. + ActivityDefinition struct { + // Name identifies the activity within its backend registry. + Name string + // Options supplies portable registration defaults. + Options ActivityOptions + // Handler must be func(context.Context, I) (O, error) or func(context.Context, I) error. + // Use Activity[I, O].Register for compile-time checking of input and output. + Handler any + } + + // ActivityCall carries a typed input unchanged to the backend's serializer. + ActivityCall struct { + // Name identifies a registered activity. + Name string + // Input is passed to the backend codec without a wrapper envelope. + Input any + // Options overrides registration defaults according to backend rules. + Options ActivityOptions + } + + // ActivityFuture is the backend result boundary. Application code should use + // Activity[I, O], which exposes Future[O] instead of result pointers. + ActivityFuture interface { + // Get waits using the captured workflow scope and decodes into a pointer. + Get(context.Context, any) error + // IsReady reports completion without blocking. + IsReady() bool + } + + // Activity binds a stable activity name to its input and output types. + // Reusing the same value for registration and invocation keeps both ends typed. + Activity[I, O any] struct { + // Name must be stable across worker restarts and workflow replay. + Name string + } + + typedActivityFuture[O any] struct { + future ActivityFuture + } +) + +// Register installs the typed handler and its default scheduling options. +func (a Activity[I, O]) Register(ctx context.Context, registrar ActivityRegistrar, opts ActivityOptions, fn func(context.Context, I) (O, error)) error { + return registrar.RegisterActivity(ctx, ActivityDefinition{Name: a.Name, Options: opts, Handler: fn}) +} + +// Execute schedules the activity and waits for its typed result. +func (a Activity[I, O]) Execute(ctx context.Context, executor ActivityExecutor, input I, opts ActivityOptions) (O, error) { + var out O + err := executor.ExecuteActivity(ctx, ActivityCall{Name: a.Name, Input: input, Options: opts}, &out) + return out, err +} + +// ExecuteAsync schedules the activity without waiting for completion. +func (a Activity[I, O]) ExecuteAsync(ctx context.Context, executor ActivityExecutor, input I, opts ActivityOptions) (Future[O], error) { + future, err := executor.ExecuteActivityAsync(ctx, ActivityCall{Name: a.Name, Input: input, Options: opts}) + if err != nil { + return nil, err + } + return typedActivityFuture[O]{future: future}, nil +} + +func (f typedActivityFuture[O]) Get(ctx context.Context) (O, error) { + var out O + err := f.future.Get(ctx, &out) + return out, err +} + +func (f typedActivityFuture[O]) IsReady() bool { + return f.future.IsReady() +} diff --git a/execution/adapter/handler.go b/execution/adapter/handler.go new file mode 100644 index 000000000..e51a93903 --- /dev/null +++ b/execution/adapter/handler.go @@ -0,0 +1,89 @@ +// Package adapter implements the type-erased adapter boundary for typed activities. +package adapter + +import ( + "context" + "errors" + "fmt" + "reflect" +) + +// Handler retains the concrete function signature required by backend codecs. +type Handler struct { + fn reflect.Value +} + +// New validates a handler once, before backend registration. +func New(fn any) (Handler, error) { + v := reflect.ValueOf(fn) + if !v.IsValid() || v.Kind() != reflect.Func || v.IsNil() { + return Handler{}, errors.New("activity handler is required") + } + t := v.Type() + if t.IsVariadic() || t.NumIn() != 2 || t.In(0) != reflect.TypeFor[context.Context]() || (t.NumOut() != 1 && t.NumOut() != 2) || t.Out(t.NumOut()-1) != reflect.TypeFor[error]() { + return Handler{}, errors.New("activity handler must be func(context.Context, I) (O, error) or func(context.Context, I) error") + } + return Handler{fn: v}, nil +} + +// Wrap preserves the native signature while adapting context and errors. +func (h Handler) Wrap(ctxFn func(context.Context) context.Context, errFn func(error) error) any { + return reflect.MakeFunc(h.fn.Type(), func(args []reflect.Value) []reflect.Value { + args[0] = reflect.ValueOf(ctxFn(args[0].Interface().(context.Context))) + out := h.fn.Call(args) + last := len(out) - 1 + if !out[last].IsNil() { + err := errFn(out[last].Interface().(error)) + out[last] = reflect.ValueOf(&err).Elem() + } + return out + }).Interface() +} + +// Call invokes a validated handler without a serialization round trip. +func (h Handler) Call(ctx context.Context, input any) (any, error) { + v := reflect.ValueOf(input) + if !v.IsValid() || !v.Type().AssignableTo(h.fn.Type().In(1)) { + return nil, fmt.Errorf("activity input must have type %s", h.fn.Type().In(1)) + } + out := h.fn.Call([]reflect.Value{reflect.ValueOf(ctx), v}) + var result any = struct{}{} + if len(out) == 2 { + result = out[0].Interface() + } + if err := out[len(out)-1]; !err.IsNil() { + return result, err.Interface().(error) + } + return result, nil +} + +// Assign copies a result into its typed destination without a serialization round trip. +func Assign(dst, src any) error { + v := reflect.ValueOf(dst) + if !v.IsValid() || v.Kind() != reflect.Pointer || v.IsNil() { + return errors.New("activity result must be a non-nil pointer") + } + out := reflect.ValueOf(src) + // A nil interface has no reflect type; typed nil pointers and maps do. + if !out.IsValid() && v.Elem().Kind() == reflect.Interface { + v.Elem().SetZero() + return nil + } + if !out.IsValid() || !out.Type().AssignableTo(v.Elem().Type()) { + return fmt.Errorf("activity result cannot be assigned to %s", v.Elem().Type()) + } + v.Elem().Set(out) + return nil +} + +// ValidateCall rejects malformed invocations before scheduling commands. +func ValidateCall(name string, input any) error { + if name == "" { + return errors.New("activity name is required") + } + v := reflect.ValueOf(input) + if !v.IsValid() || (v.Kind() == reflect.Pointer && v.IsNil()) { + return errors.New("activity input is required") + } + return nil +} diff --git a/execution/options.go b/execution/options.go new file mode 100644 index 000000000..6254f52da --- /dev/null +++ b/execution/options.go @@ -0,0 +1,52 @@ +package execution + +import ( + "context" + "time" +) + +type ( + // Future represents a pending activity result. It belongs to the workflow + // execution that scheduled it and must not be shared across workflows. + // Waiting and cancellation follow the selected backend's rules. + Future[T any] interface { + // Get blocks until the activity completes and returns the typed result. + // Once completed, repeated calls return the recorded result. + Get(ctx context.Context) (T, error) + + // IsReady returns true if the activity has completed (success or failure) and Get() + // will not block. This allows workflows to poll or implement custom waiting strategies. + IsReady() bool + } + + // ActivityOptions configures retry and timeouts for an activity. + ActivityOptions struct { + // Queue overrides the activity queue. Empty uses registration or backend defaults. + Queue string + // RetryPolicy controls retry behavior for this activity. If zero-valued, the + // engine uses its default retry policy. + RetryPolicy RetryPolicy + // ScheduleToStartTimeout bounds how long the activity may wait in the task + // queue before a worker starts the attempt. Zero means leave queue-wait + // unspecified here and let the engine adapter apply its own defaults. + ScheduleToStartTimeout time.Duration + // StartToCloseTimeout bounds one activity attempt once a worker has started + // executing it. Zero means use the engine default. + StartToCloseTimeout time.Duration + // HeartbeatTimeout bounds the maximum gap between heartbeats emitted by the + // running activity. Zero leaves the backend default unchanged. + HeartbeatTimeout time.Duration + } + + // RetryPolicy defines retry semantics shared by workflows and activities. + // Zero-valued fields mean the engine uses its defaults. + RetryPolicy struct { + // MaxAttempts caps total attempts, including the first. Zero uses backend defaults. + MaxAttempts int + // InitialInterval is the delay before the first retry. Zero means use engine default. + InitialInterval time.Duration + // BackoffCoefficient multiplies the delay after each retry. Zero uses backend + // defaults; the backend defines the valid range of explicit coefficients. + BackoffCoefficient float64 + } +) diff --git a/execution/selvedge/activities.go b/execution/selvedge/activities.go new file mode 100644 index 000000000..57ea95674 --- /dev/null +++ b/execution/selvedge/activities.go @@ -0,0 +1,118 @@ +// Package selvedge binds execution activities to native Selvedge apps and +// workflow scopes. Applications retain ownership of app lifecycle, workflows, +// submission, transactions, and backend-specific policies. +package selvedge + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/CaliLuke/loom/execution" + "github.com/CaliLuke/loom/execution/adapter" + sv "github.com/CaliLuke/selvedge" +) + +type ( + // Activities registers named handlers on one caller-owned Selvedge app. + // Register all handlers before starting the app so recovery can find them. + Activities struct { + app *sv.App + mu sync.RWMutex + definitions map[string]invocation + } + + invocation func(*sv.WorkflowContext, any, execution.ActivityOptions) (execution.ActivityFuture, error) + + scope struct { + activities *Activities + ctx *sv.WorkflowContext + } + + result[O any] struct { + ctx *sv.WorkflowContext + future *sv.Future[O] + } +) + +// New binds activity registration to app without starting or stopping it. +func New(app *sv.App) (*Activities, error) { + if app == nil { + return nil, errors.New("selvedge: activity registration requires an app") + } + return &Activities{app: app, definitions: make(map[string]invocation)}, nil +} + +// Register installs a typed activity with native Selvedge options and codecs. +// Its generic signature preserves the input and output types for Selvedge; +// Activities therefore does not implement execution.ActivityRegistrar. +func Register[I, O any](activities *Activities, activity execution.Activity[I, O], handler func(context.Context, I) (O, error), options sv.ActivityOptions) error { + if activity.Name == "" { + return errors.New("selvedge: activity name is required") + } + if handler == nil { + return errors.New("selvedge: activity handler is required") + } + activities.mu.Lock() + defer activities.mu.Unlock() + if _, exists := activities.definitions[activity.Name]; exists { + return fmt.Errorf("selvedge: activity %q already registered", activity.Name) + } + definition := sv.DefineActivity(activity.Name, handler, sv.WithActivityDefaults[I, O](options)) + if err := sv.RegisterActivity(activities.app, definition); err != nil { + return err + } + activities.definitions[activity.Name] = func(ctx *sv.WorkflowContext, input any, opts execution.ActivityOptions) (execution.ActivityFuture, error) { + typed, ok := input.(I) + if !ok { + return nil, fmt.Errorf("selvedge: activity %q input must have type %T", activity.Name, *new(I)) + } + future, err := definition.ExecuteAsync(ctx, typed, sv.WithActivityOptions(invocationActivityOptions(opts))) + if err != nil { + return nil, err + } + return result[O]{ctx: ctx, future: future}, nil + } + return nil +} + +// Scope binds execution to ctx, including its cancellation and detached-scope +// behavior. The context.Context arguments of execution calls do not replace this +// native workflow context. Futures retain the scope that scheduled them. +func (a *Activities) Scope(ctx *sv.WorkflowContext) execution.ActivityExecutor { + return scope{activities: a, ctx: ctx} +} + +func (s scope) ExecuteActivity(ctx context.Context, call execution.ActivityCall, out any) error { + future, err := s.ExecuteActivityAsync(ctx, call) + if err != nil { + return err + } + return future.Get(ctx, out) +} + +func (s scope) ExecuteActivityAsync(_ context.Context, call execution.ActivityCall) (execution.ActivityFuture, error) { + if err := adapter.ValidateCall(call.Name, call.Input); err != nil { + return nil, err + } + s.activities.mu.RLock() + invoke := s.activities.definitions[call.Name] + s.activities.mu.RUnlock() + if invoke == nil { + return nil, fmt.Errorf("selvedge: activity %q is not registered", call.Name) + } + return invoke(s.ctx, call.Input, call.Options) +} + +func (r result[O]) Get(_ context.Context, out any) error { + value, err := r.future.Get(r.ctx) + if err != nil { + return err + } + return adapter.Assign(out, value) +} + +func (r result[O]) IsReady() bool { + return r.future.IsReady() +} diff --git a/execution/selvedge/activities_test.go b/execution/selvedge/activities_test.go new file mode 100644 index 000000000..7215b5510 --- /dev/null +++ b/execution/selvedge/activities_test.go @@ -0,0 +1,100 @@ +package selvedge_test + +import ( + "context" + "errors" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/CaliLuke/loom/execution" + adapter "github.com/CaliLuke/loom/execution/selvedge" + sv "github.com/CaliLuke/selvedge" + "github.com/CaliLuke/selvedge/retry" + "github.com/stretchr/testify/require" +) + +type metadata struct { + Key string `json:"key"` +} + +func TestNativePolicyAndTypedResults(t *testing.T) { + app := newApp(t) + activities, err := adapter.New(app) + require.NoError(t, err) + activity := execution.Activity[metadata, string]{Name: "metadata.read"} + var attempts atomic.Int32 + require.NoError(t, adapter.Register(activities, activity, func(_ context.Context, input metadata) (string, error) { + if attempts.Add(1) == 1 { + return "", errors.New("temporary object store failure") + } + return input.Key, nil + }, sv.ActivityOptions{StartToCloseTimeout: time.Second, RetryPolicy: retry.Fixed(2, time.Millisecond)})) + workflow := sv.DefineWorkflow("metadata.refresh", func(ctx *sv.WorkflowContext, input metadata) (string, error) { + future, err := activity.ExecuteAsync(context.Background(), activities.Scope(ctx), input, execution.ActivityOptions{}) + if err != nil { + return "", err + } + out, err := future.Get(context.Background()) + if err != nil { + return "", err + } + again, err := future.Get(context.Background()) + if err == nil && (again != out || !future.IsReady()) { + return "", errors.New("completed future changed") + } + return out, err + }) + sv.RegisterWorkflow(app, workflow) + ctx := startApp(t, app) + id, err := sv.StartWorkflow(ctx, app, workflow, metadata{Key: "asset.png"}) + require.NoError(t, err) + result, err := sv.WaitWorkflow[string](ctx, app, id) + require.NoError(t, err) + require.NoError(t, result.Error) + require.Equal(t, "asset.png", result.Output) + require.EqualValues(t, 2, attempts.Load()) +} + +func TestNativePermanentErrorPolicy(t *testing.T) { + app := newApp(t) + activities, err := adapter.New(app) + require.NoError(t, err) + activity := execution.Activity[metadata, string]{Name: "metadata.read"} + permanent := errors.New("permanent") + policy := retry.Fixed(3, time.Millisecond) + policy.NonRetryableErrors = []error{permanent} + var attempts atomic.Int32 + require.NoError(t, adapter.Register(activities, activity, func(context.Context, metadata) (string, error) { + attempts.Add(1) + return "", permanent + }, sv.ActivityOptions{StartToCloseTimeout: time.Second, RetryPolicy: policy})) + workflow := sv.DefineWorkflow("metadata.refresh", func(ctx *sv.WorkflowContext, input metadata) (string, error) { + return activity.Execute(context.Background(), activities.Scope(ctx), input, execution.ActivityOptions{}) + }) + sv.RegisterWorkflow(app, workflow) + ctx := startApp(t, app) + id, err := sv.StartWorkflow(ctx, app, workflow, metadata{Key: "asset.png"}) + require.NoError(t, err) + result, err := sv.WaitWorkflow[string](ctx, app, id) + require.NoError(t, err) + require.Error(t, result.Error) + require.EqualValues(t, 1, attempts.Load()) +} + +func newApp(t *testing.T) *sv.App { + t.Helper() + return sv.NewApp(sv.WithDatabase("file:"+filepath.Join(t.TempDir(), "workflow.db")), sv.WithWaitPollInterval(time.Millisecond), sv.WithActivityPollInterval(time.Millisecond)) +} + +func startApp(t *testing.T, app *sv.App) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + require.NoError(t, app.Start(ctx)) + t.Cleanup(func() { + require.NoError(t, app.Shutdown(context.Background())) + }) + return ctx +} diff --git a/execution/selvedge/activity_options_test.go b/execution/selvedge/activity_options_test.go new file mode 100644 index 000000000..1e08abccc --- /dev/null +++ b/execution/selvedge/activity_options_test.go @@ -0,0 +1,37 @@ +package selvedge_test + +import ( + "context" + "testing" + "time" + + "github.com/CaliLuke/loom/execution" + adapter "github.com/CaliLuke/loom/execution/selvedge" + sv "github.com/CaliLuke/selvedge" + "github.com/CaliLuke/selvedge/retry" + "github.com/stretchr/testify/require" +) + +func TestPortableRetryOverridePreservesNativeMaximumInterval(t *testing.T) { + app := newApp(t) + activities, err := adapter.New(app) + require.NoError(t, err) + activity := execution.Activity[metadata, time.Duration]{Name: "metadata.read"} + policy := retry.Exponential(3, time.Millisecond, 7*time.Second, 2) + require.NoError(t, adapter.Register(activities, activity, func(ctx context.Context, _ metadata) (time.Duration, error) { + return sv.GetActivityInfo(ctx).Options.RetryPolicy.MaxInterval, nil + }, sv.ActivityOptions{StartToCloseTimeout: time.Second, RetryPolicy: policy})) + workflow := sv.DefineWorkflow("metadata.refresh", func(ctx *sv.WorkflowContext, input metadata) (time.Duration, error) { + return activity.Execute(ctx.Context(), activities.Scope(ctx), input, execution.ActivityOptions{ + RetryPolicy: execution.RetryPolicy{InitialInterval: time.Second}, + }) + }) + sv.RegisterWorkflow(app, workflow) + ctx := startApp(t, app) + id, err := sv.StartWorkflow(ctx, app, workflow, metadata{Key: "asset.png"}) + require.NoError(t, err) + result, err := sv.WaitWorkflow[time.Duration](ctx, app, id) + require.NoError(t, err) + require.NoError(t, result.Error) + require.Equal(t, 7*time.Second, result.Output) +} diff --git a/execution/selvedge/options.go b/execution/selvedge/options.go new file mode 100644 index 000000000..7c9460497 --- /dev/null +++ b/execution/selvedge/options.go @@ -0,0 +1,17 @@ +package selvedge + +import ( + "github.com/CaliLuke/loom/execution" + sv "github.com/CaliLuke/selvedge" + "github.com/CaliLuke/selvedge/retry" +) + +// invocationActivityOptions leaves zero fields unset so recorded definition defaults survive. +func invocationActivityOptions(options execution.ActivityOptions) sv.ActivityOptions { + result := sv.ActivityOptions{Queue: options.Queue, StartToCloseTimeout: options.StartToCloseTimeout, ScheduleToStartTimeout: options.ScheduleToStartTimeout, HeartbeatTimeout: options.HeartbeatTimeout} + p := options.RetryPolicy + if p.MaxAttempts != 0 || p.InitialInterval != 0 || p.BackoffCoefficient != 0 { + result.RetryPolicy = &retry.Policy{MaxAttempts: p.MaxAttempts, InitialInterval: p.InitialInterval, Multiplier: p.BackoffCoefficient} + } + return result +} diff --git a/execution/selvedge/registration_test.go b/execution/selvedge/registration_test.go new file mode 100644 index 000000000..9aae258f0 --- /dev/null +++ b/execution/selvedge/registration_test.go @@ -0,0 +1,60 @@ +package selvedge_test + +import ( + "context" + "testing" + + "github.com/CaliLuke/loom/execution" + adapter "github.com/CaliLuke/loom/execution/selvedge" + sv "github.com/CaliLuke/selvedge" + "github.com/stretchr/testify/require" +) + +func TestRegistrationRejectsInvalidHandlersAndDuplicateNames(t *testing.T) { + _, err := adapter.New(nil) + require.Error(t, err) + activities, err := adapter.New(newApp(t)) + require.NoError(t, err) + var nilHandler func(context.Context, metadata) (string, error) + require.Error(t, adapter.Register(activities, execution.Activity[metadata, string]{Name: "metadata.read"}, nilHandler, sv.ActivityOptions{})) + activity := execution.Activity[metadata, string]{Name: "metadata.read"} + handler := func(_ context.Context, input metadata) (string, error) { + return input.Key, nil + } + require.Error(t, adapter.Register(activities, execution.Activity[metadata, string]{}, handler, sv.ActivityOptions{})) + require.NoError(t, adapter.Register(activities, activity, handler, sv.ActivityOptions{})) + require.Error(t, adapter.Register(activities, activity, handler, sv.ActivityOptions{})) +} + +func TestNativeConflictDoesNotPublishLocalDefinition(t *testing.T) { + app := newApp(t) + native := sv.DefineActivity("metadata.read", func(context.Context, metadata) (string, error) { + return "native", nil + }) + require.NoError(t, sv.RegisterActivity(app, native)) + activities, err := adapter.New(app) + require.NoError(t, err) + activity := execution.Activity[metadata, string]{Name: "metadata.read"} + err = adapter.Register(activities, activity, func(context.Context, metadata) (string, error) { + return "conflicting", nil + }, sv.ActivityOptions{}) + require.ErrorIs(t, err, sv.ErrActivityNameConflict) + workflow := sv.DefineWorkflow("probe", func(ctx *sv.WorkflowContext, input metadata) (string, error) { + _, err := activity.Execute(context.Background(), activities.Scope(ctx), input, execution.ActivityOptions{}) + if err == nil { + return "published", nil + } + return "rejected", nil + }) + sv.RegisterWorkflow(app, workflow) + ctx := startApp(t, app) + id, err := sv.StartWorkflow(ctx, app, workflow, metadata{Key: "asset.png"}) + require.NoError(t, err) + result, err := sv.WaitWorkflow[string](ctx, app, id) + require.NoError(t, err) + require.NoError(t, result.Error) + require.Equal(t, "rejected", result.Output) + tasks, err := sv.ListActivityTasks(ctx, app, id) + require.NoError(t, err) + require.Empty(t, tasks) +} diff --git a/execution/selvedge/scope_test.go b/execution/selvedge/scope_test.go new file mode 100644 index 000000000..5c63f2763 --- /dev/null +++ b/execution/selvedge/scope_test.go @@ -0,0 +1,50 @@ +package selvedge_test + +import ( + "context" + "errors" + "testing" + + "github.com/CaliLuke/loom/execution" + adapter "github.com/CaliLuke/loom/execution/selvedge" + sv "github.com/CaliLuke/selvedge" + "github.com/stretchr/testify/require" +) + +func TestActivityScopesStayIndependent(t *testing.T) { + app := newApp(t) + activities, err := adapter.New(app) + require.NoError(t, err) + activity := execution.Activity[metadata, string]{Name: "metadata.read"} + require.NoError(t, adapter.Register(activities, activity, func(_ context.Context, input metadata) (string, error) { + return input.Key, nil + }, sv.ActivityOptions{})) + workflow := sv.DefineWorkflow("scopes", func(ctx *sv.WorkflowContext, input metadata) (string, error) { + parent, err := activity.ExecuteAsync(ctx.Context(), activities.Scope(ctx), input, execution.ActivityOptions{}) + if err != nil { + return "", err + } + child, cancel := sv.WithCancel(ctx) + cancel() + if _, err := activity.Execute(ctx.Context(), activities.Scope(child), input, execution.ActivityOptions{}); !errors.Is(err, context.Canceled) { + return "", errors.New("canceled scope admitted an activity") + } + detached, err := activity.Execute(ctx.Context(), activities.Scope(sv.Detached(child)), input, execution.ActivityOptions{}) + if err != nil { + return "", err + } + out, err := parent.Get(ctx.Context()) + if err == nil && out != detached { + return "", errors.New("activity scopes changed results") + } + return out, err + }) + sv.RegisterWorkflow(app, workflow) + ctx := startApp(t, app) + id, err := sv.StartWorkflow(ctx, app, workflow, metadata{Key: "asset.png"}) + require.NoError(t, err) + result, err := sv.WaitWorkflow[string](ctx, app, id) + require.NoError(t, err) + require.NoError(t, result.Error) + require.Equal(t, "asset.png", result.Output) +} diff --git a/execution/selvedge/tla/ActivityBinding.tla b/execution/selvedge/tla/ActivityBinding.tla new file mode 100644 index 000000000..780b99c11 --- /dev/null +++ b/execution/selvedge/tla/ActivityBinding.tla @@ -0,0 +1,54 @@ +-------------------------- MODULE ActivityBinding -------------------------- +EXTENDS Naturals, FiniteSets +CONSTANTS Names, Scopes, RebindFuture, PublishConflict +VARIABLES native, owned, registered, futures, nextID +vars == <> + +Init == /\ native = {} + /\ owned = {} + /\ registered = {} + /\ futures = {} + /\ nextID = 0 + +\* An independently registered native activity can conflict with our registry. +NativeRegister(name) == + /\ native' = native \cup {name} + /\ UNCHANGED <> + +\* The registry lock spans native registration and local publication. A native +\* conflict must not publish a definition that workers would never execute. +\* owned records successful adapter registrations, excluding native conflicts. +\* PublishConflict models publication despite a failed native registration. +Register(name) == + /\ name \notin registered + /\ (name \notin native \/ PublishConflict) + /\ native' = native \cup {name} + /\ owned' = IF name \notin native THEN owned \cup {name} ELSE owned + /\ registered' = registered \cup {name} + /\ UNCHANGED <> + +Schedule(name, scope) == + /\ name \in registered + /\ nextID < 2 + /\ futures' = futures \cup {[id |-> nextID, name |-> name, + scheduled |-> scope, awaited |-> scope]} + /\ nextID' = nextID + 1 + /\ UNCHANGED <> + +\* Counterexample: Get uses a later caller's workflow context instead of the +\* scope captured when the command was scheduled. +Get(future, scope) == + /\ future \in futures + /\ futures' = (futures \ {future}) \cup + {[future EXCEPT !.awaited = IF RebindFuture THEN scope ELSE @]} + /\ UNCHANGED <> + +Next == \/ \E name \in Names: NativeRegister(name) \/ Register(name) + \/ \E name \in Names, scope \in Scopes: Schedule(name, scope) + \/ \E future \in futures, scope \in Scopes: Get(future, scope) + +PublishedDefinitionsExist == registered \subseteq owned /\ owned \subseteq native +FutureScopeIsStable == \A future \in futures: + future.scheduled = future.awaited /\ future.name \in registered +Spec == Init /\ [][Next]_vars +============================================================================= diff --git a/execution/selvedge/tla/README.md b/execution/selvedge/tla/README.md new file mode 100644 index 000000000..e8804f88a --- /dev/null +++ b/execution/selvedge/tla/README.md @@ -0,0 +1,51 @@ +# Activity registration and workflow-scope binding + +`ActivityBinding.tla` checks the adapter boundary extracted into +`execution/selvedge`. Registration publishes a local definition only after native +registration succeeds, and futures keep the workflow scope that scheduled them. +The `owned` set records successful adapter registrations separately from all +native names. Every published definition must belong to that set. +The registry lock spans both registration steps. A scope is an immutable value; +there is no registry-wide "current workflow" context. + +Run with TLA+ tools v1.7.4 (TLC 2.19), Java 25, from this directory: + +```sh +java -XX:+UseParallelGC -cp /path/to/tla2tools-1.7.4.jar tlc2.TLC \ + -workers 1 -deadlock -metadir /tmp/activity-binding-checked \ + -config checked.cfg ActivityBinding.tla +java -XX:+UseParallelGC -cp /path/to/tla2tools-1.7.4.jar tlc2.TLC \ + -workers 1 -deadlock -metadir /tmp/activity-binding-rebound \ + -config rebound.cfg ActivityBinding.tla +java -XX:+UseParallelGC -cp /path/to/tla2tools-1.7.4.jar tlc2.TLC \ + -workers 1 -deadlock -metadir /tmp/activity-binding-conflict \ + -config conflict.cfg ActivityBinding.tla +``` + +The checked configuration passed: 53 distinct states, 321 generated states. +The rebound configuration must fail `FutureScopeIsStable`: it models an +alternative that selects the caller's later scope during `Get`, moving a future +from its parent scope to a detached scope. TLC found that four-state trace. +The conflict configuration must fail `PublishedDefinitionsExist`: it publishes +a local definition after an independent native registration claims the same +name. TLC found that three-state trace, with no adapter-owned native definition. +Both negative configurations exit with status 12; the checked one exits with 0. + +Before tracking ownership, the invariant checked only that every published name +existed in the native registry. Removing the native-conflict guard still passed +that weaker invariant (53 distinct states, 409 generated states). The ownership +invariant and conflict configuration close that proof gap; runtime registration +already rejected this case. + +Correspondence tests are `TestNativeConflictDoesNotPublishLocalDefinition` and +`TestActivityScopesStayIndependent`. Agent conformance and saved workflow histories check the shared `execution` +contracts with agent-owned serializers, error conversion and defaults. The +consuming application's metadata tests exercise this native Selvedge binding. + +Bounds: two names, two scopes, two futures. Native registration is assumed to +reject conflicts without replacing the installed handler. Selvedge owns command +ordering, retries, persistence, transactions, and cancellation implementation; +this model does not prove those backend properties. Native transaction admission +is retained and tested in the consuming application rather than reimplemented +in this adapter. The model is a design check, not a claim that the extracted +implementation previously had a scope-rebinding defect. diff --git a/execution/selvedge/tla/checked.cfg b/execution/selvedge/tla/checked.cfg new file mode 100644 index 000000000..976ca758d --- /dev/null +++ b/execution/selvedge/tla/checked.cfg @@ -0,0 +1,7 @@ +CONSTANTS +Names = {a, b} +Scopes = {parent, detached} +RebindFuture = FALSE +PublishConflict = FALSE +SPECIFICATION Spec +INVARIANTS PublishedDefinitionsExist FutureScopeIsStable diff --git a/execution/selvedge/tla/conflict.cfg b/execution/selvedge/tla/conflict.cfg new file mode 100644 index 000000000..4cb756ff2 --- /dev/null +++ b/execution/selvedge/tla/conflict.cfg @@ -0,0 +1,7 @@ +CONSTANTS +Names = {a, b} +Scopes = {parent, detached} +RebindFuture = FALSE +PublishConflict = TRUE +SPECIFICATION Spec +INVARIANTS PublishedDefinitionsExist FutureScopeIsStable diff --git a/execution/selvedge/tla/rebound.cfg b/execution/selvedge/tla/rebound.cfg new file mode 100644 index 000000000..64ace345e --- /dev/null +++ b/execution/selvedge/tla/rebound.cfg @@ -0,0 +1,7 @@ +CONSTANTS +Names = {a, b} +Scopes = {parent, detached} +RebindFuture = TRUE +PublishConflict = FALSE +SPECIFICATION Spec +INVARIANTS PublishedDefinitionsExist FutureScopeIsStable diff --git a/go.mod b/go.mod index 1c438d60c..bfa2c902b 100644 --- a/go.mod +++ b/go.mod @@ -1,11 +1,12 @@ module github.com/CaliLuke/loom -go 1.27.2 +go 1.27.3 require ( github.com/CaliLuke/go-redis/v9 v9.23.1-0.20261014134601-1073297e76ea github.com/CaliLuke/go-sse v0.11.1-0.20261008010800-b29b33ae1a1e github.com/CaliLuke/libopenapi v0.41.7-0.20261014145029-7886f0170cc4 + github.com/CaliLuke/selvedge v0.7.0 github.com/alecthomas/kong v1.16.1 github.com/alicebob/miniredis/v2 v2.40.0 github.com/aws/smithy-go v1.28.4 @@ -51,12 +52,20 @@ require ( github.com/buger/jsonparser v1.6.1 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.31.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/pgx/v5 v5.11.0 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/manveru/gobdd v0.0.0-20131210092515-f1a17fdd710b // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/ncruces/go-strftime v0.1.9 // indirect github.com/pb33f/jsonpath v0.8.4 // indirect github.com/pb33f/ordered-map/v2 v2.3.2 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect github.com/yuin/gopher-lua v1.1.2 // indirect @@ -64,7 +73,12 @@ require ( go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.47.0 // indirect go.opentelemetry.io/proto/otlp v1.11.1 // indirect go.uber.org/atomic v1.12.0 // indirect + golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba // indirect golang.org/x/sync v0.24.0 // indirect golang.org/x/sys v0.49.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20261005182115-fad411399dd8 // indirect + modernc.org/libc v1.66.10 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.11.0 // indirect + modernc.org/sqlite v1.40.1 // indirect ) diff --git a/go.sum b/go.sum index f43d3ae4c..a08394131 100644 --- a/go.sum +++ b/go.sum @@ -1,9 +1,17 @@ +dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= +dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= +github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0= +github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= github.com/CaliLuke/go-redis/v9 v9.23.1-0.20261014134601-1073297e76ea h1:oRvCGHzzNf4TdpFJJxgvx5wzBGRwW8QStIkUC04A1bI= github.com/CaliLuke/go-redis/v9 v9.23.1-0.20261014134601-1073297e76ea/go.mod h1:Uvij0aiDtA9pejtqMT84jqWSTEE+OYsph2DRDybsGhs= github.com/CaliLuke/go-sse v0.11.1-0.20261008010800-b29b33ae1a1e h1:8ikgsAY/zoRUd/7kLCrBjEVTKxNc99TA647E+Jcke8k= github.com/CaliLuke/go-sse v0.11.1-0.20261008010800-b29b33ae1a1e/go.mod h1:lXdhhI7K/QSTmuy5OiNyu32AQ/bUZ8aE8wj8taCn3QE= github.com/CaliLuke/libopenapi v0.41.7-0.20261014145029-7886f0170cc4 h1:KG8QSbAtOfHv2JatFv9XsqI673y6tNPMWmpc0VN59y4= github.com/CaliLuke/libopenapi v0.41.7-0.20261014145029-7886f0170cc4/go.mod h1:gEQ2ydf5xr92Ql+uRl54gig6N/DAc/zuIrfZNqtFqPA= +github.com/CaliLuke/selvedge v0.7.0 h1:9scL+JWLODAOie5C6vHKkkSvknyLWglUQT5c2Ij6YM8= +github.com/CaliLuke/selvedge v0.7.0/go.mod h1:WMOxrwKeto/YPgLNhH9mxvduVEiWlVKDGLkm2uAUZw8= +github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= +github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/alecthomas/assert/v2 v2.11.0 h1:2Q9r3ki8+JYXvGsDyBXwH3LcJ+WK5D0gc5E8vS6K3D0= github.com/alecthomas/assert/v2 v2.11.0/go.mod h1:Bze95FyfUr7x34QZrjL+XP+0qgp/zg8yS+TtBj1WA3k= github.com/alecthomas/kong v1.16.1 h1:ixhCt93XkJ98kGposQ54+bl0IK6XwqB40AsMynU7Z8E= @@ -22,15 +30,39 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= github.com/buger/jsonparser v1.6.1 h1:I0phFv0PlbLHnM7TZAVjZ2MJ2/eWRTDyuO7GLR98IEs= github.com/buger/jsonparser v1.6.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= +github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= +github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= +github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= +github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= +github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= +github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A= +github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw= +github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA= +github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= github.com/dave/jennifer v1.7.1 h1:B4jJJDHelWcDhlRQxWeo0Npa/pYKBLrirAQoTN45txo= github.com/dave/jennifer v1.7.1/go.mod h1:nXbxhEmQfOZhWml3D1cDK5M1FLnMSozpbFN/m3RmGZc= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dimfeld/httppath v0.0.0-20170720192232-ee938bf73598 h1:MGKhKyiYrvMDZsmLR/+RGffQSXwEkXgfLSA08qDn9AI= github.com/dimfeld/httppath v0.0.0-20170720192232-ee938bf73598/go.mod h1:0FpDmbrt36utu8jEmeU05dPC9AB5tsLYVVi+ZHfyuwI= +github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= +github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/docker/docker v28.5.1+incompatible h1:Bm8DchhSD2J6PsFzxC35TZo4TLGR2PdW/E69rU45NhM= +github.com/docker/docker v28.5.1+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= +github.com/docker/go-connections v0.6.0 h1:LlMG9azAe1TqfR7sO+NJttz1gy6KO7VJBh+pMmjSD94= +github.com/docker/go-connections v0.6.0/go.mod h1:AahvXYshr6JgfUJGdDCs2b5EZG/vmaMAntpSFH5BFKE= +github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= +github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/ebitengine/purego v0.8.4 h1:CF7LEKg5FFOsASUj0+QwaXf8Ht6TlFxg09+S9wz0omw= +github.com/ebitengine/purego v0.8.4/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeORc= github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE= github.com/go-chi/chi/v5 v5.3.2 h1:5YQkICvTCSZ25hoRsyJazN0scjzKGiu4VAUc7H1o1nY= @@ -40,10 +72,14 @@ github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8= github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= @@ -52,14 +88,52 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.31.0 h1:Bd7KaOxzULLxtZ/K5s1aLbWhR0+ github.com/grpc-ecosystem/grpc-gateway/v2 v2.31.0/go.mod h1:nN7ts3dFXKtCZWc//yfkpcQNKJABg16/uDVAZpLDalo= github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM= github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= +github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/klauspost/cpuid/v2 v2.4.0 h1:S6Hrbc7+ywsr0r+RLapfGBHfyefhCTwEh3A0tV913Dw= github.com/klauspost/cpuid/v2 v2.4.0/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= +github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= +github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= github.com/manveru/faker v0.0.0-20171103152722-9fbc68a78c4d h1:Zj+PHjnhRYWBK6RqCDBcAhLXoi3TzC27Zad/Vn+gnVQ= github.com/manveru/faker v0.0.0-20171103152722-9fbc68a78c4d/go.mod h1:WZy8Q5coAB1zhY9AOBJP0O6J4BuDfbupUDavKY+I3+s= github.com/manveru/gobdd v0.0.0-20131210092515-f1a17fdd710b h1:3E44bLeN8uKYdfQqVQycPnaVviZdBLbizFhU49mtbe4= github.com/manveru/gobdd v0.0.0-20131210092515-f1a17fdd710b/go.mod h1:Bj8LjjP0ReT1eKt5QlKjwgi5AFm5mI6O1A2G4ChI0Ag= +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/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= +github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= +github.com/moby/go-archive v0.1.0 h1:Kk/5rdW/g+H8NHdJW2gsXyZ7UnzvJNOy6VKJqueWdcQ= +github.com/moby/go-archive v0.1.0/go.mod h1:G9B+YoujNohJmrIYFBpSd54GTUB4lt9S+xVQvsJyFuo= +github.com/moby/patternmatcher v0.6.0 h1:GmP9lR19aU5GqSSFko+5pRqHi+Ohk1O69aFiKkVGiPk= +github.com/moby/patternmatcher v0.6.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc= +github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU= +github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko= +github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= +github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs= +github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g= +github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28= +github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0= +github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y= +github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= +github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= +github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= +github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/oklog/ulid/v2 v2.1.2 h1:IEclFb9JNvzYA6MW2SCxbLzcHTVsfqm3PrqGQJH5zec= github.com/oklog/ulid/v2 v2.1.2/go.mod h1:rcEKHmBBKfef9DhnvX7y1HZBYxjXb0cP5ExxNsTT1QQ= +github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= +github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= +github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= +github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= github.com/owenrumney/go-sarif/v3 v3.3.1 h1:bKY8YnWcY8zYTiGVObC5wZLlfEvMkCxK9XMGXBbzEtQ= github.com/owenrumney/go-sarif/v3 v3.3.1/go.mod h1:72MaugkExDexbSauRuPq6BvUAAqAX0TwoNYMIQyZCMw= github.com/pb33f/go-yaml v0.1.1 h1:yWGmhVdwzionRnSc/hEJwcfq6PaZL2QdYzyReXBbX7Q= @@ -71,13 +145,30 @@ github.com/pb33f/ordered-map/v2 v2.3.2/go.mod h1:1OhFrXu3OYw3kM+FXF+Ug3ugmmZ9LE8 github.com/pb33f/testify v0.1.1 h1:mnHe7uxKt8dyNYEGUspow72VjD264DB5rOJq4bz5tJw= github.com/pb33f/testify v0.1.1/go.mod h1:keghMqOLECF1ENzESnfM+cwPcKN5uhTOpvbtjgL6aZg= github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o= +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/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw= +github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +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/shirou/gopsutil/v4 v4.25.6 h1:kLysI2JsKorfaFPcYmcJqbzROzsBWEOAtw6A7dIfqXs= +github.com/shirou/gopsutil/v4 v4.25.6/go.mod h1:PfybzyydfZcN+JMMjkF6Zb8Mq1A/VcogFFg7hj50W9c= github.com/sirupsen/logrus v1.10.2 h1:G2SED73/qrAu6YwbdxOD6peLkCBI3z7L+ykJFTXJBBo= github.com/sirupsen/logrus v1.10.2/go.mod h1:SLEg8TqYulVKKfIGHldVp2K2aYz2DKSVBq4g/H5bR7Q= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= +github.com/testcontainers/testcontainers-go v0.40.0 h1:pSdJYLOVgLE8YdUY2FHQ1Fxu+aMnb6JfVz1mxk7OeMU= +github.com/testcontainers/testcontainers-go v0.40.0/go.mod h1:FSXV5KQtX2HAMlm7U3APNyLkkap35zNLxukw9oBi/MY= +github.com/testcontainers/testcontainers-go/modules/postgres v0.40.0 h1:s2bIayFXlbDFexo96y+htn7FzuhpXLYJNnIuglNKqOk= +github.com/testcontainers/testcontainers-go/modules/postgres v0.40.0/go.mod h1:h+u/2KoREGTnTl9UwrQ/g+XhasAT8E6dClclAADeXoQ= +github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU= +github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI= +github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk= +github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY= github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU= github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb h1:zGWFAtiMcyryUHoUjUJX0/lt1H2+i2Ka2n+D3DImSNo= github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU= @@ -87,6 +178,8 @@ github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y= github.com/yuin/gopher-lua v1.1.2 h1:yF/FjE3hD65tBbt0VXLE13HWS9h34fdzJmrWRXwobGA= github.com/yuin/gopher-lua v1.1.2/go.mod h1:7aRmXIWl37SqRf0koeyylBEzJ+aPt8A+mmkQ4f1ntR8= +github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= +github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= @@ -133,12 +226,17 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= +golang.org/x/crypto v0.58.0 h1:COkYLr4k7nDI4r6QtMML2AfzeIswWHAJthLS6K/M/54= +golang.org/x/crypto v0.58.0/go.mod h1:Xh+kl5A+M0gMiWDytH2Hhr5Z7WbNCpv/eB/EFgfSk+w= +golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba h1:Ck8QetSgk912qxWLMCKxd0in+aiyBQyDSMae6e/xmpU= +golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba/go.mod h1:50RgIsmK7OwqzTTeqcSXQW8SswW0o8fRcDxmqGluJ8E= golang.org/x/mod v0.42.0 h1:ICeL3t2L30ljSPHHxDnR4fqVcBt/zR2zJAb9juavmQg= golang.org/x/mod v0.42.0/go.mod h1:u5fzIrRInkL8MaO1cbRkUHT64v303YIT5KEqpNr7jRs= golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw= golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI= golang.org/x/sync v0.24.0 h1:iqiDrGoDgyuvAKrhLQthUEzU/R378N78gXbBjxed7oA= golang.org/x/sync v0.24.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA= golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.47.0 h1:cpyOwQkr2Nzxe2p6KnrTG7Ei2wtzjsuSjhpXXepZ/BM= @@ -157,3 +255,31 @@ google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0= google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +modernc.org/cc/v4 v4.26.5 h1:xM3bX7Mve6G8K8b+T11ReenJOT+BmVqQj0FY5T4+5Y4= +modernc.org/cc/v4 v4.26.5/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0= +modernc.org/ccgo/v4 v4.28.1 h1:wPKYn5EC/mYTqBO373jKjvX2n+3+aK7+sICCv4Fjy1A= +modernc.org/ccgo/v4 v4.28.1/go.mod h1:uD+4RnfrVgE6ec9NGguUNdhqzNIeeomeXf6CL0GTE5Q= +modernc.org/fileutil v1.3.40 h1:ZGMswMNc9JOCrcrakF1HrvmergNLAmxOPjizirpfqBA= +modernc.org/fileutil v1.3.40/go.mod h1:HxmghZSZVAz/LXcMNwZPA/DRrQZEVP9VX0V4LQGQFOc= +modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= +modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= +modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= +modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= +modernc.org/libc v1.66.10 h1:yZkb3YeLx4oynyR+iUsXsybsX4Ubx7MQlSYEw4yj59A= +modernc.org/libc v1.66.10/go.mod h1:8vGSEwvoUoltr4dlywvHqjtAqHBaw0j1jI7iFBTAr2I= +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.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= +modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8= +modernc.org/opt v0.1.4/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.40.1 h1:VfuXcxcUWWKRBuP8+BR9L7VnmusMgBNNnBYGEe9w/iY= +modernc.org/sqlite v1.40.1/go.mod h1:9fjQZ0mB1LLP0GYrp39oOJXx/I2sxEnZtzCmEQIKvGE= +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= diff --git a/grpc/integration_tests/fixtures/quality/go.mod b/grpc/integration_tests/fixtures/quality/go.mod index 04b2b3097..7aeb79c7b 100644 --- a/grpc/integration_tests/fixtures/quality/go.mod +++ b/grpc/integration_tests/fixtures/quality/go.mod @@ -1,6 +1,6 @@ module example.com/grpc-quality -go 1.27.2 +go 1.27.3 require ( github.com/CaliLuke/loom v1.7.1 diff --git a/http/integration_tests/fixtures/quality/go.mod b/http/integration_tests/fixtures/quality/go.mod index 652dd9bb8..4f1ab027a 100644 --- a/http/integration_tests/fixtures/quality/go.mod +++ b/http/integration_tests/fixtures/quality/go.mod @@ -1,6 +1,6 @@ module example.com/http-quality -go 1.27.2 +go 1.27.3 require github.com/CaliLuke/loom v1.7.1 diff --git a/http/integration_tests/fixtures/ticktock/go.mod b/http/integration_tests/fixtures/ticktock/go.mod index 282ff3d5f..a5992bd02 100644 --- a/http/integration_tests/fixtures/ticktock/go.mod +++ b/http/integration_tests/fixtures/ticktock/go.mod @@ -1,6 +1,6 @@ module example.com/http-ticktock -go 1.27.2 +go 1.27.3 require ( github.com/CaliLuke/loom v1.7.1 diff --git a/http/integration_tests/go.mod b/http/integration_tests/go.mod index 8b5b96a56..e494c2686 100644 --- a/http/integration_tests/go.mod +++ b/http/integration_tests/go.mod @@ -1,6 +1,6 @@ module github.com/CaliLuke/loom/http/integration_tests -go 1.27.2 +go 1.27.3 require ( github.com/CaliLuke/loom v1.7.1 diff --git a/http/testdata/form_comparison/go.mod b/http/testdata/form_comparison/go.mod index a4db97f50..77c10d540 100644 --- a/http/testdata/form_comparison/go.mod +++ b/http/testdata/form_comparison/go.mod @@ -1,6 +1,6 @@ module github.com/CaliLuke/loom/http/testdata/form_comparison -go 1.27.2 +go 1.27.3 require ( github.com/CaliLuke/loom v1.10.0-alpha.4.0.20261008072015-bd83a10dfccc diff --git a/jsonrpc/codegen/envelope_id_readiness_test.go b/jsonrpc/codegen/envelope_id_readiness_test.go index 485732e45..a03aa0413 100644 --- a/jsonrpc/codegen/envelope_id_readiness_test.go +++ b/jsonrpc/codegen/envelope_id_readiness_test.go @@ -28,7 +28,7 @@ func TestJSONRPCEnvelopeIDMounted(t *testing.T) { dir = output require.NoError(t, os.MkdirAll(dir, 0o700)) } - require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte(fmt.Sprintf("module example.com/r4id\n\ngo 1.27.2\n\nrequire github.com/CaliLuke/loom v1.0.0\nreplace github.com/CaliLuke/loom => %s\n", resolveJSONRPCTestLoomModulePath(t, dir))), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte(fmt.Sprintf("module example.com/r4id\n\ngo 1.27.3\n\nrequire github.com/CaliLuke/loom v1.0.0\nreplace github.com/CaliLuke/loom => %s\n", resolveJSONRPCTestLoomModulePath(t, dir))), 0o600)) require.NoError(t, os.MkdirAll(filepath.Join(dir, "design"), 0o700)) require.NoError(t, os.WriteFile(filepath.Join(dir, "design", "design.go"), []byte(`package design import "github.com/CaliLuke/loom/jsonrpc/codegen/testdata" diff --git a/jsonrpc/integration_tests/fixtures/mixedtick/go.mod b/jsonrpc/integration_tests/fixtures/mixedtick/go.mod index 5fc52c229..dc85b6bb7 100644 --- a/jsonrpc/integration_tests/fixtures/mixedtick/go.mod +++ b/jsonrpc/integration_tests/fixtures/mixedtick/go.mod @@ -1,6 +1,6 @@ module example.com/mixedtick -go 1.27.2 +go 1.27.3 require github.com/CaliLuke/loom v1.7.1 diff --git a/jsonrpc/integration_tests/fixtures/ticktock/go.mod b/jsonrpc/integration_tests/fixtures/ticktock/go.mod index d3e45369f..9f2d06339 100644 --- a/jsonrpc/integration_tests/fixtures/ticktock/go.mod +++ b/jsonrpc/integration_tests/fixtures/ticktock/go.mod @@ -1,6 +1,6 @@ module example.com/ticktock -go 1.27.2 +go 1.27.3 require ( github.com/CaliLuke/loom v1.7.1 diff --git a/jsonrpc/integration_tests/go.mod b/jsonrpc/integration_tests/go.mod index 190c023c6..d1a8ebe68 100644 --- a/jsonrpc/integration_tests/go.mod +++ b/jsonrpc/integration_tests/go.mod @@ -1,6 +1,6 @@ module github.com/CaliLuke/loom/jsonrpc/integration_tests -go 1.27.2 +go 1.27.3 require ( github.com/CaliLuke/loom v1.7.1 diff --git a/pulse/rmap/replica_recovery_test.go b/pulse/rmap/replica_recovery_test.go index a53fd5f25..9bf4a1daf 100644 --- a/pulse/rmap/replica_recovery_test.go +++ b/pulse/rmap/replica_recovery_test.go @@ -1,13 +1,11 @@ package rmap import ( - "context" "fmt" "maps" "testing" "time" - "github.com/CaliLuke/go-redis/v9" "github.com/stretchr/testify/require" "github.com/CaliLuke/loom/pulse/internal/redistest" @@ -155,36 +153,16 @@ func TestReplicaNoOpPreservesStateAndTTL(t *testing.T) { } } -type retireWriteHook struct { - retire func(context.Context) error -} - -func (h retireWriteHook) DialHook(next redis.DialHook) redis.DialHook { - return next -} -func (h retireWriteHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook { - return next -} -func (h retireWriteHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook { - return func(ctx context.Context, cmd redis.Cmder) error { - if err := next(ctx, cmd); err != nil { - return err - } - if cmd.Name() == "evalsha" && cmd.Args()[1] == luaSet.Hash() { - return h.retire(ctx) - } - return nil - } -} - func TestReplicaWriteObservationRejectsRetiredGeneration(t *testing.T) { srv := redistest.Start(t, redistest.DBRmap) - m := joinMap(t, srv.Client, "retire-before-observe") - require.NoError(t, m.sub.Unsubscribe(t.Context(), m.chankey)) - srv.Client.AddHook(retireWriteHook{retire: func(ctx context.Context) error { - return srv.Client.Del(ctx, m.hashkey).Err() - }}) - _, err := m.SetAndWait(t.Context(), "lost", "value") + writer := joinMap(t, srv.Client, "retire-before-observe") + committed, err := writer.runSetScript(t.Context(), "lost", "value") + require.NoError(t, err) + // Isolate this observer from Pub/Sub and periodic repair: either could + // legitimately observe the committed revision before it is retired. + m := newPendingObservationMap(t, writer) + require.NoError(t, srv.Client.Del(t.Context(), m.hashkey).Err()) + err = m.observe(observationRequest{ctx: t.Context(), generation: committed.generation, rev: committed.rev}) require.ErrorContains(t, err, "retired before observation") require.Empty(t, m.Map()) }