diff --git a/design/README.md b/design/README.md index 70c9d48..bdc56a1 100644 --- a/design/README.md +++ b/design/README.md @@ -27,6 +27,7 @@ When a document diverges significantly from the code, it lists a "Gap" section i - [observability.md](observability.md) — minimal runtime observability facts and performance bounds. - [benchmarks.md](benchmarks.md) — what the performance baseline measures and does not, and the measurement conditions comparable numbers must carry. - [api-documentation.md](api-documentation.md) — English doc comments for the public API: contract boundaries, scope, example trade-offs, and the acceptance process. +- [conformance-example.md](conformance-example.md) — the Single Silo public API conformance flow: application records, pending actions, Reminder recovery, and release evidence. - [release.md](release.md) — version numbers, release thresholds, the manual release checklist, and how release-note blocks are handled. - [release-0.1.0.md](release-0.1.0.md) — the implementation order, failure matrix, conformance example, and evidence gates for the first announced release. diff --git a/design/conformance-example.md b/design/conformance-example.md new file mode 100644 index 0000000..6ce9bc6 --- /dev/null +++ b/design/conformance-example.md @@ -0,0 +1,344 @@ +# Public API conformance example + +## Decision + +Extend `examples/shadow` with a small recovery path. `Device` remains the +main Grain. Add a fixed-key `RecoveryCoordinator` Grain and an +Application-owned store for the recovery records. + +The conformance path is Single Silo. It must not configure a `MemberStore`, a +Transport, or cluster options. The existing `net`-tagged cluster test is a +separate preview test. It is not part of this conformance path. + +The example uses only public gor APIs for Runtime work. It does not read the +Runtime's private state, call a private runtime method, or repair a Runtime +database row. + +## Grain shape + +The example keeps the existing `Device` Grain and adds these conformance +methods to its typed interface. Existing device-shadow methods remain +available. + +```go +type Device interface { + Report(ctx context.Context, workshopID string, state string) error + ReportAction(ctx context.Context, actionID string, state string) error + Shadow(ctx context.Context) (Shadow, error) + ShadowExists(ctx context.Context) (bool, error) + ClearShadow(ctx context.Context) error + ApplyPending(ctx context.Context, actionID string) error +} + +type RecoveryCoordinator interface { + Start(ctx context.Context) error + Stop(ctx context.Context) error + Recover(ctx context.Context, tick gor.TickStatus) error +} +``` + +The interfaces have the `//gor:grain` marker. The generated package supplies +the typed Grain References, dispatch functions, and Reminder call factory. +Startup calls the generated `Install` function before `Register` or `Ref`. + +`RecoveryCoordinatorKey` is the only key used for the coordinator. Its value +is a fixed application constant, such as `"recovery"`. A device uses its +device key as its `GrainKey`. The coordinator calls each target through +`gor.Ref[Device](b, action.DeviceKey)`. This proves a typed cross-Grain Call +without adding cluster ownership to the example. + +## Data ownership + +The example must keep Runtime data and Application data in separate stores. + +| Data | Owner | Boundary and contract | +| --- | --- | --- | +| Device shadow | Grain Runtime | A named `gor.State[Shadow]` called `"shadow"`. `Set`, `Exists`, `Get`, and `Clear` use the public State API. | +| Coordinator status | Grain Runtime | A named `gor.State[CoordinatorState]` called `"running"`. It is a restart-visible status value, not the recovery queue. | +| Recovery schedule | Grain Runtime | A `gor.Reminder[RecoveryCoordinator]` called `"recovery"`, set with `gor.Every` and `gor.Handle(RecoveryCoordinator.Recover)`. | +| Pending actions and applied records | Application | An application-defined `ApplicationStore` interface. Its SQLite implementation uses a separate `business.db` file and its own tables. | + +The Runtime State store and Runtime Reminder store use the existing public +`store.Store` and `store.ReminderStore` interfaces. The ApplicationStore is +not a new gor feature. The Application owns its interface, schema, migration, +transaction, and close operation. Tests use an in-memory implementation with +fault injection. + +The application store must not use a `store.GrainId` row as a disguised +business record. It must not inspect the Runtime's `records` or `schedule` +tables. A normal run opens two durable paths: + +```text +runtime.db Runtime Reminder and membership tables +runtime-state.db Runtime Grain State created by store.OpenSQLite +business.db Application pending actions and applied records +``` + +The example runs without membership, so the membership table is unused. The +ApplicationStore owns `business.db`; the Runtime owns the other two files. + +## Application records + +The ApplicationStore has the smallest interface needed by the two Grains: + +```go +type ApplicationStore interface { + SavePending(context.Context, PendingAction) error + ListPending(context.Context) ([]PendingAction, error) + ApplyPending(context.Context, string) error + ReadApplied(context.Context, string) (AppliedRecord, bool, error) + Close() error +} +``` + +`PendingAction` contains an application-generated `ActionID`, the target +device key, the reported value, and an optional trace ID. `AppliedRecord` is a +business receipt keyed by `ActionID`. The ActionID is the deduplication key; +the example never generates a new ActionID while retrying an uncertain +action. + +`SavePending` inserts a new action. Repeating the same ActionID with the same +payload is a no-op. Reusing an ActionID with a different payload returns an +application error. `ListPending` returns only pending actions in a stable +ActionID order. + +`ApplyPending` is one application transaction. It reads the action, creates +one applied receipt, and marks the action applied. A unique ActionID makes the +transaction safe to repeat. If the action is already applied, the operation +returns success without creating another receipt or changing the business +result. + +The Runtime does not know these records and does not make this transaction +atomic with a State or Reminder write. The separate boundary is the reason +the ActionID and Safe Repeat rule are required. + +## Call flow + +### Start recovery + +The application opens both stores, creates a Single Silo, installs generated +bindings, and registers both Grain factories. The factories capture the +ApplicationStore. They still receive the public `*gor.Binder` and create all +Runtime State and Reminder handles from that Binder. + +The first call is: + +```go +gor.Ref[RecoveryCoordinator](rt, RecoveryCoordinatorKey).Start(ctx) +``` + +`Start` sets the `"recovery"` Reminder to a short periodic interval and then +confirms the `"running"` State. If `"running"` is already present and true, +`Start` does nothing. This makes an optional startup check safe after a +restart; it does not reset the Reminder's first tick time. + +`Stop` clears the `"running"` State first and then cancels the Reminder by +name. A successful cancel removes the persisted Reminder. If the process stops +after State clear and before cancel, a restart sees a stopped coordinator and +`Start` safely restores the Reminder. The example calls `Stop` only in the +cancellation test; recovery after a process stop does not call `Stop` or +rewrite the schedule. + +### Save a pending action + +The caller adds Request Context and calls one typed Device Reference: + +```go +ctx, err := gor.WithRequestContext(context.Background(), "trace_id", "trace-1") +if err != nil { + return err +} +err = gor.Ref[Device](rt, "device-1").ReportAction(ctx, "report-1", "temperature=20") +``` + +`Device.ReportAction` performs these steps in order: + +1. Read `trace_id` with `gor.RequestContextValue` and validate it. +2. Save `PendingAction{ActionID: "report-1", ...}` through ApplicationStore. +3. Read the current shadow State. +4. Write the new shadow with `State.Set`. + +The pending save comes before the State write so a deterministic +`ErrPendingActionConflict` leaves Device State unchanged. This is not a +distributed transaction: if a pending save succeeds but the later State write +has an uncertain result, retry the same ActionID. The copied trace ID is +application data. It is not Request Context after the write. Request Context +is not stored in Runtime State or Reminder records. +The `Recover` method must observe an empty Request Context because the +Runtime creates Reminder Calls with a fresh context. + +`ShadowExists` calls `State.Exists`. `ClearShadow` calls `State.Clear`. These +methods make absence observable and prove that an absent State is different +from a present zero value. + +### Recover after a stop + +The deterministic restart test calls `rt.Kill()` after `Report` returns and +before the next Reminder delivery. It then creates a new Runtime with the +same public stores, installs the same generated bindings, and advances the +injected clock. No private runtime call and no database repair is allowed. + +The persisted Reminder poller does this for each due row: + +1. List the due Reminder. +2. Claim it with the public `ReminderStore.Claim` CAS. +3. Build `TickStatus` and invoke `RecoveryCoordinator.Recover` as an ordinary + typed Grain Call. + +`Recover` lists pending Application actions in ActionID order. For each one it +calls `Device.ApplyPending`. The Device delegates to the ApplicationStore +transaction. A successful recovery leaves one applied record and no pending +record for that ActionID. + +The test also wraps the public `ReminderStore` in a test-only adapter that +blocks after a successful `Claim`. The test driver calls public `rt.Kill` +while the claim is blocked, then releases the adapter. This proves the +claim-before-delivery boundary: the claimed due time can be missed, the +Device method does not run, and the pending application record remains. A +periodic Reminder gives the Application a later recovery attempt. The test +does not insert or edit a Reminder row by hand. + +### Safe Repeat + +The Runtime promises at-most-once Reminder delivery for each claimed due +time. It does not promise exactly-once execution. A process stop after a claim +can lose that delivery. A caller timeout, cancellation, or unknown store +result can also leave the caller unable to know whether a Business Action ran. + +The recovery method therefore uses this Safe Repeat rule: + +> For one ActionID, `ApplyPending` may run any number of times, but the +> Application transaction may create one applied receipt only. + +The repeat test invokes `ApplyPending` twice through the typed Device +Reference. A second call returns success and does not create a second receipt. +The stronger fault test commits the first application transaction and then +returns an injected unknown error. The error reaches the Reminder error sink; +the same ActionID is called again and the receipt count remains one. + +The Runtime does not add a retry. The periodic Reminder and the Application's +ActionID rule provide recovery. + +## Failures and Unknown Results + +The example always installs both `gor.OnError` and `gor.OnCall`. + +- A foreground Device or coordinator Call returns its error to the caller. + `OnCall` records the method, duration, and error for the test. +- A failed `Recover` method has no waiting caller. `OnError` receives the + original error with `ReminderInvocation{Method: "Recover"}`. +- A failed `OnDeactivate` hook is reported with `gor.Deactivation`. +- Reminder scan and claim failures are scheduler failures. They do not enter + `OnError`; the test observes that no Device Call ran and that a failed claim + leaves the row available. The next poll retries the scan. +- A failed Reminder method is not retried for the same due time. The + application may recover the pending ActionID on a later periodic tick. + +The example documents these intentional Unknown Results: + +| Boundary | Unknown result | Required handling | +| --- | --- | --- | +| Device `State.Set` or `State.Clear` | A non-context store error can mean that the write committed or did not commit. The activation is discarded. | Call again to load confirmed State. Retry a Business Action only with the same ActionID and a Safe Repeat rule. | +| `ApplicationStore.SavePending` | The pending row can exist even when `ReportAction` returns an error. | Read by ActionID before creating another action. Retry the same ActionID. | +| `ApplicationStore.ApplyPending` | The application transaction can commit before its result reaches the Grain. | Retry the same ActionID. The unique receipt makes the retry a no-op. | +| `Reminder.Set` or `Reminder.Cancel` | The unconditional write or delete can be complete when the caller sees an error. | Repeat Set by the same name or repeat Cancel. Do not edit Runtime tables. | +| Call timeout or cancellation | The caller stopped waiting; the Grain method may have started and may have saved State or an action. | Treat the result as unknown. Query the application record and use the same ActionID before retry. | +| Process stop after Reminder claim | The due occurrence may be missed because claim happens before delivery. | Leave the pending action in ApplicationStore and wait for the next periodic Reminder. | + +The Single Silo conformance path has no Transport, so it does not produce a +transport failure. The public call contract still treats a transport failure +as unknown when a clustered caller uses the API outside this example. + +The example must not turn any row count, error, or timeout into a false +success. Every error from store open, Runtime creation, registration, a Call, +Reminder setup, ApplicationStore, or close is returned or reported. The only +errors not sent to `OnError` are the scheduler failures and shutdown +cancellations defined by the public Reminder contract. + +## Tests and evidence + +The conformance tests use `clock.NewFake`, `testing/synctest`, public Memory +stores, and test-owned storage adapters. They must not use sleeps, wall-clock +polling, private Runtime fields, or manual SQL repair. + +The minimum test cases are: + +1. Install generated bindings, register both Grain types, obtain typed + References, and call the Device and fixed-key coordinator. +2. Set shadow State, assert `Exists`, clear it, and assert absence after + reactivation. Also test a present zero value separately from absence. +3. Add Request Context, copy its value deliberately into Application data, + and prove it is absent from Runtime State, Reminder data, and a Reminder + Call. +4. Save an action, stop the Runtime before delivery, restart with the same + stores, and recover the action without repair. +5. Run two concurrent public `Claim` attempts and assert one winner. A + successful claim must precede the Device Call. +6. Stop after a successful claim and before delivery. Assert that the action + remains pending and that the next periodic tick recovers it. +7. Fail before an application commit and assert `OnError`, one later retry, + and one applied receipt. +8. Commit then return an injected unknown result. Retry the same ActionID and + assert one applied receipt. +9. Invoke the same ApplyPending action twice and assert Safe Repeat behavior. +10. Cancel the recovery Reminder, clear coordinator State, and assert that no + later tick runs. + +The existing State, ReminderStore, Request Context, error-sink, and Runtime +restart tests remain lower-level evidence. The conformance tests prove their +composition. A test is not green because it only requested a run; it must +assert the observed Call, stored record, error source, and nonzero test count. + +## Clean module and release gates + +The generated file is committed. A clean consumer check must build the +committed example without access to the repository's build cache, then run +the two process phases against a temporary pair of database paths: + +```bash +consumer_dir="$(mktemp -d)" +mod_cache="$consumer_dir/modcache" +(cd "$consumer_dir" && go mod init conformance-check) +GOMODCACHE="$mod_cache" go run github.com/suraciii/gor/examples/shadow/cmd/conformance@v0.1.0 \ + -phase prepare -db "$consumer_dir/runtime.db" -business-db "$consumer_dir/business.db" +GOMODCACHE="$mod_cache" go run github.com/suraciii/gor/examples/shadow/cmd/conformance@v0.1.0 \ + -phase recover -db "$consumer_dir/runtime.db" -business-db "$consumer_dir/business.db" +``` + +The prepare phase saves the action and exits with Reminder polling disabled. +The recover phase starts a new process with polling enabled and waits for the +observed recovery Call. It then reads the application receipt and exits with a +nonzero status if the receipt is missing or duplicated. The release check +uses the candidate version in place of `v0.1.0` before the tag exists. + +The release candidate must pass every existing gate: + +```text +make test +make sim +make gen +make net +make lint +go test -count=1 -race ./... +make ci +``` + +`make ci` is the required aggregate gate. It includes format checking, lint, +the default tests, race tests, simulation, generated-code tests, and network +tests. The clean consumer build and the two-phase run are additional release +evidence; they do not replace `make ci`. + +## API decision and Gap + +No new framework feature is required. Public `State`, `Reminder`, typed Grain +References, Request Context, `OnError`, `OnCall`, `Kill`, `New`, `Store`, and +`ReminderStore` provide all Runtime seams used here. The ApplicationStore is +application code because the Runtime must not own or interpret business +records. + +The coordinator, ApplicationStore, process driver, generated bindings, and +conformance tests are implemented in `examples/shadow`. The existing shadow +`Device.Report` keeps its workshopID contract, so the conformance path uses the +separate typed `ReportAction` method for an ActionID. This keeps workshopID and +ActionID separate while preserving the Single Silo boundary. The example does +not configure cluster membership, a Transport, or private Runtime state. diff --git a/design/release-0.1.0.md b/design/release-0.1.0.md index bf6a857..d8935d0 100644 --- a/design/release-0.1.0.md +++ b/design/release-0.1.0.md @@ -118,7 +118,8 @@ Add a small example and tests that use only public APIs. It must contain: The example is not a second framework. It proves that the public Runtime boundaries support a real durable application pattern with local application -data and safe repeat handling. +data and safe repeat handling. The required Single Silo flow and its failure +evidence are specified in [conformance-example.md](conformance-example.md). ### 6. Run release checks @@ -168,7 +169,9 @@ The first example is the main usability test. A user must be able to: 8. close and reopen the Runtime. Each step must have one clear public path. The example must not need cache -details, private store layout, or a second hidden retry loop. +details, private store layout, cluster membership, or a second hidden retry +loop. The clean consumer build and two-process run are additional evidence +for the conformance example. ## Work order diff --git a/docs/release-0.1.0.md b/docs/release-0.1.0.md index 7fe1732..60a42ab 100644 --- a/docs/release-0.1.0.md +++ b/docs/release-0.1.0.md @@ -149,6 +149,8 @@ The release is ready only when all items below are true: ## Gap -The single-Silo Runtime, State, Reminders, typed Calls, lifecycle, and -observations already exist in parts. The 0.1.0 work is not complete until -these parts use the public Grain model together under restart and failure. +The conformance Application in `examples/shadow` now composes the +single-Silo Runtime, State, Reminders, typed Calls, Request Context, lifecycle, +and observations under restart and failure. It keeps business records in a +separate ApplicationStore and uses ActionID Safe Repeat. The remaining release +status is tracked in ROADMAP.md and the release gates. diff --git a/examples/shadow/cmd/conformance/main.go b/examples/shadow/cmd/conformance/main.go new file mode 100644 index 0000000..034d9a6 --- /dev/null +++ b/examples/shadow/cmd/conformance/main.go @@ -0,0 +1,294 @@ +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log" + "os" + "path/filepath" + "strings" + "time" + + "github.com/suraciii/gor" + shadow "github.com/suraciii/gor/examples/shadow" + "github.com/suraciii/gor/examples/shadow/domain" + "github.com/suraciii/gor/store" +) + +const ( + phasePrepare = "prepare" + phaseRecover = "recover" +) + +func main() { + if err := run(context.Background(), os.Args[1:]); err != nil { + log.Fatal(err) + } +} + +func run(ctx context.Context, args []string) (runErr error) { + flags := flag.NewFlagSet("conformance", flag.ContinueOnError) + flags.SetOutput(os.Stderr) + phase := flags.String("phase", "", "process phase: prepare or recover") + runtimePath := flags.String("db", "runtime.db", "Runtime SQLite database path") + businessPath := flags.String("business-db", "business.db", "Application SQLite database path") + deviceKey := flags.String("device", "device-1", "target Device GrainKey") + actionID := flags.String("action-id", "action-1", "pending Business ActionID") + state := flags.String("state", "temperature=20", "reported Device State") + traceID := flags.String("trace-id", "trace-1", "Request Context trace_id for prepare") + waitTimeout := flags.Duration("timeout", 10*time.Second, "maximum wait for the recovery Call") + if err := flags.Parse(args); err != nil { + return err + } + if flags.NArg() != 0 { + return fmt.Errorf("unexpected arguments: %v", flags.Args()) + } + if *phase != phasePrepare && *phase != phaseRecover { + return fmt.Errorf("-phase must be %q or %q", phasePrepare, phaseRecover) + } + if *waitTimeout <= 0 { + return errors.New("-timeout must be positive") + } + if err := validateDatabasePaths(*runtimePath, *businessPath); err != nil { + return err + } + if err := makeParent(*runtimePath); err != nil { + return fmt.Errorf("create Runtime database directory: %w", err) + } + if err := makeParent(*businessPath); err != nil { + return fmt.Errorf("create Application database directory: %w", err) + } + + runtimeStore, err := store.OpenSQLite(*runtimePath) + if err != nil { + return fmt.Errorf("open Runtime database: %w", err) + } + application, err := domain.OpenSQLiteApplicationStore(*businessPath) + if err != nil { + runtimeStore.Close() + return fmt.Errorf("open Application database: %w", err) + } + defer func() { + if err := application.Close(); err != nil { + runErr = errors.Join(runErr, fmt.Errorf("close Application database: %w", err)) + } + if err := runtimeStore.Close(); err != nil { + runErr = errors.Join(runErr, fmt.Errorf("close Runtime database: %w", err)) + } + }() + + calls := make(chan gor.CallObservation, 32) + options := []gor.Option{ + gor.WithStore(runtimeStore), + gor.WithReminderStore(runtimeStore), + gor.WithReminderInterval(0), + gor.OnError(shadow.LogBackgroundError), + gor.OnCall(func(observation gor.CallObservation) { calls <- observation }), + } + if *phase == phaseRecover { + options = append(options, gor.WithReminderInterval(domain.RecoveryInterval)) + } + rt, err := gor.New(options...) + if err != nil { + return fmt.Errorf("create Single Silo Runtime: %w", err) + } + defer rt.Close() + if err := shadow.RegisterConformance(rt, application); err != nil { + return fmt.Errorf("register conformance Grains: %w", err) + } + + coordinator := gor.Ref[domain.RecoveryCoordinator](rt, domain.RecoveryCoordinatorKey) + switch *phase { + case phasePrepare: + if err := coordinator.Start(ctx); err != nil { + return fmt.Errorf("start recovery coordinator: %w", err) + } + requestContext, err := gor.WithRequestContext(ctx, "trace_id", *traceID) + if err != nil { + return fmt.Errorf("add Request Context: %w", err) + } + if err := gor.Ref[domain.Device](rt, *deviceKey).ReportAction(requestContext, *actionID, *state); err != nil { + return fmt.Errorf("save pending action: %w", err) + } + log.Printf("prepared ActionID %q for Device %q; stop the process before Reminder delivery", *actionID, *deviceKey) + return nil + case phaseRecover: + if err := coordinator.Start(ctx); err != nil { + return fmt.Errorf("start recovery coordinator: %w", err) + } + if err := waitForRecovery(ctx, calls, *waitTimeout); err != nil { + return err + } + record, applied, err := application.ReadApplied(ctx, *actionID) + if err != nil { + return fmt.Errorf("read applied record: %w", err) + } + if !applied { + return fmt.Errorf("ActionID %q has no applied record", *actionID) + } + pending, err := application.ListPending(ctx) + if err != nil { + return fmt.Errorf("list pending actions: %w", err) + } + if len(pending) != 0 { + return fmt.Errorf("pending actions remain after recovery: %#v", pending) + } + log.Printf("recovered ActionID %q for Device %q with receipt %#v", record.ActionID, record.DeviceKey, record) + return nil + } + return nil +} + +func waitForRecovery(ctx context.Context, calls <-chan gor.CallObservation, timeout time.Duration) error { + waitContext, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + for { + select { + case observation := <-calls: + if observation.Method != "Recover" { + continue + } + if observation.Err != nil { + return fmt.Errorf("recovery Call failed: %w", observation.Err) + } + return nil + case <-waitContext.Done(): + return fmt.Errorf("wait for recovery Call: %w", waitContext.Err()) + } + } +} + +type databasePathFamily struct { + label string + members []string +} + +func validateDatabasePaths(runtimePath, businessPath string) error { + bases := []struct { + label string + path string + }{ + {label: "runtime coordination", path: runtimePath}, + {label: "runtime state", path: derivedRuntimeStatePath(runtimePath)}, + {label: "application", path: businessPath}, + } + families := make([]databasePathFamily, len(bases)) + for index, base := range bases { + absolute, err := cleanDatabasePath(base.path) + if err != nil { + return fmt.Errorf("resolve %s database path: %w", base.label, err) + } + members := []string{absolute, absolute + "-wal", absolute + "-shm"} + for _, member := range members[1:] { + if err := rejectSymlinkComponents(member); err != nil { + return fmt.Errorf("resolve %s database path: %w", base.label, err) + } + } + families[index] = databasePathFamily{label: base.label, members: members} + } + var paths []struct { + label string + path string + } + for _, family := range families { + for _, member := range family.members { + paths = append(paths, struct { + label string + path string + }{label: family.label, path: member}) + } + } + for left := 0; left < len(paths); left++ { + for right := left + 1; right < len(paths); right++ { + if paths[left].path == paths[right].path { + return fmt.Errorf("database paths for %s and %s must be different: both resolve to %q", paths[left].label, paths[right].label, paths[left].path) + } + same, err := sameExistingFile(paths[left].path, paths[right].path) + if err != nil { + return fmt.Errorf("compare %s and %s database paths: %w", paths[left].label, paths[right].label, err) + } + if same { + return fmt.Errorf("database paths for %s and %s must not alias an existing file", paths[left].label, paths[right].label) + } + } + } + return nil +} + +func derivedRuntimeStatePath(runtimePath string) string { + directory, base := filepath.Split(runtimePath) + extension := filepath.Ext(base) + return filepath.Join(directory, strings.TrimSuffix(base, extension)+"-state"+extension) +} + +func cleanDatabasePath(path string) (string, error) { + if path == "" { + return "", errors.New("database path is empty") + } + for _, component := range strings.Split(filepath.ToSlash(path), "/") { + if component == ".." { + return "", errors.New("database path must not contain '..'") + } + } + absolute, err := filepath.Abs(path) + if err != nil { + return "", err + } + if err := rejectSymlinkComponents(absolute); err != nil { + return "", err + } + return filepath.Clean(absolute), nil +} + +func rejectSymlinkComponents(absolute string) error { + volume := filepath.VolumeName(absolute) + rest := strings.TrimPrefix(absolute, volume) + current := volume + separator := string(filepath.Separator) + if strings.HasPrefix(rest, separator) { + current = volume + separator + rest = strings.TrimPrefix(rest, separator) + } + for _, component := range strings.Split(rest, separator) { + if component == "" || component == "." { + continue + } + current = filepath.Join(current, component) + info, err := os.Lstat(current) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return err + } + if info.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("database path contains symlink component %q", current) + } + } + return nil +} + +func sameExistingFile(leftPath, rightPath string) (bool, error) { + left, leftErr := os.Stat(leftPath) + if leftErr != nil && !errors.Is(leftErr, os.ErrNotExist) { + return false, leftErr + } + right, rightErr := os.Stat(rightPath) + if rightErr != nil && !errors.Is(rightErr, os.ErrNotExist) { + return false, rightErr + } + if leftErr != nil || rightErr != nil { + return false, nil + } + return os.SameFile(left, right), nil +} + +func makeParent(path string) error { + parent := filepath.Dir(path) + if parent == "." { + return nil + } + return os.MkdirAll(parent, 0o755) +} diff --git a/examples/shadow/cmd/conformance/main_test.go b/examples/shadow/cmd/conformance/main_test.go new file mode 100644 index 0000000..9d00699 --- /dev/null +++ b/examples/shadow/cmd/conformance/main_test.go @@ -0,0 +1,170 @@ +package main + +import ( + "errors" + "os" + "path/filepath" + "strings" + "syscall" + "testing" +) + +func TestValidateDatabasePathsRejectsEquivalentPaths(t *testing.T) { + workingDir, err := os.Getwd() + if err != nil { + t.Fatal(err) + } + absolute := filepath.Join(workingDir, "runtime.db") + for _, test := range []struct { + name string + runtime string + business string + }{ + {name: "same absolute", runtime: absolute, business: absolute}, + {name: "relative and absolute", runtime: "runtime.db", business: absolute}, + {name: "cleaned relative and absolute", runtime: filepath.Join("data", "..", "runtime.db"), business: absolute}, + } { + t.Run(test.name, func(t *testing.T) { + err := validateDatabasePaths(test.runtime, test.business) + if err == nil || !strings.Contains(err.Error(), "must be different") { + t.Fatalf("validateDatabasePaths(%q, %q) = %v, want clear different-path error", test.runtime, test.business, err) + } + }) + } +} + +func TestValidateDatabasePathsAcceptsSeparatePaths(t *testing.T) { + if err := validateDatabasePaths("runtime.db", "business.db"); err != nil { + t.Fatalf("validateDatabasePaths returned error for separate paths: %v", err) + } +} + +func TestValidateDatabasePathsRejectsSQLiteSidecarAliases(t *testing.T) { + for _, sidecar := range []string{"-wal", "-shm"} { + t.Run(sidecar, func(t *testing.T) { + err := validateDatabasePaths("runtime.db", "runtime.db"+sidecar) + if err == nil || !strings.Contains(err.Error(), "must be different") { + t.Fatalf("validateDatabasePaths for %s = %v, want sidecar collision error", sidecar, err) + } + }) + } + if err := validateDatabasePaths("runtime.db", "runtime-state.db-wal"); err == nil || !strings.Contains(err.Error(), "must be different") { + t.Fatalf("validateDatabasePaths for Runtime State sidecar = %v, want sidecar collision error", err) + } +} + +func TestValidateDatabasePathsRejectsDerivedRuntimeStatePath(t *testing.T) { + directory := t.TempDir() + runtimePath := filepath.Join(directory, "runtime.db") + businessPath := derivedRuntimeStatePath(runtimePath) + if err := validateDatabasePaths(runtimePath, businessPath); err == nil || !strings.Contains(err.Error(), "runtime state") { + t.Fatalf("validateDatabasePaths(%q, %q) = %v, want derived State alias error", runtimePath, businessPath, err) + } +} + +func TestValidateDatabasePathsRejectsSymlinkAliases(t *testing.T) { + directory := t.TempDir() + runtimePath := filepath.Join(directory, "runtime.db") + businessPath := filepath.Join(directory, "business.db") + if err := os.WriteFile(runtimePath, []byte("runtime"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.Symlink(runtimePath, businessPath); err != nil { + t.Fatalf("create symlink alias: %v", err) + } + if err := validateDatabasePaths(runtimePath, businessPath); err == nil || !strings.Contains(err.Error(), "symlink") { + t.Fatalf("validateDatabasePaths(%q, %q) = %v, want symlink error", runtimePath, businessPath, err) + } + + danglingTarget := filepath.Join(directory, "not-created.db") + danglingAlias := filepath.Join(directory, "dangling-business.db") + if err := os.Symlink(danglingTarget, danglingAlias); err != nil { + t.Fatalf("create dangling symlink alias: %v", err) + } + if err := validateDatabasePaths(danglingTarget, danglingAlias); err == nil || !strings.Contains(err.Error(), "symlink") { + t.Fatalf("validateDatabasePaths(%q, %q) = %v, want dangling symlink error", danglingTarget, danglingAlias, err) + } +} + +func TestValidateDatabasePathsRejectsHardLinkAliases(t *testing.T) { + directory := t.TempDir() + runtimePath := filepath.Join(directory, "runtime.db") + businessPath := filepath.Join(directory, "business.db") + if err := os.WriteFile(runtimePath, []byte("runtime"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.Link(runtimePath, businessPath); err != nil { + if errors.Is(err, syscall.EOPNOTSUPP) || errors.Is(err, syscall.EXDEV) || errors.Is(err, syscall.EPERM) { + t.Fatalf("hard links are required for this deterministic test: %v", err) + } + t.Fatal(err) + } + if err := validateDatabasePaths(runtimePath, businessPath); err == nil || !strings.Contains(err.Error(), "must not alias") { + t.Fatalf("validateDatabasePaths(%q, %q) = %v, want hard-link alias error", runtimePath, businessPath, err) + } +} + +func TestValidateDatabasePathsRejectsSQLiteSidecarHardLinkAlias(t *testing.T) { + directory := t.TempDir() + runtimePath := filepath.Join(directory, "runtime.db") + sidecarPath := runtimePath + "-wal" + businessPath := filepath.Join(directory, "business.db") + if err := os.WriteFile(sidecarPath, []byte("runtime wal"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.Link(sidecarPath, businessPath); err != nil { + if errors.Is(err, syscall.EOPNOTSUPP) || errors.Is(err, syscall.EXDEV) || errors.Is(err, syscall.EPERM) { + t.Fatalf("hard links are required for this deterministic test: %v", err) + } + t.Fatal(err) + } + if err := validateDatabasePaths(runtimePath, businessPath); err == nil || !strings.Contains(err.Error(), "must not alias") { + t.Fatalf("validateDatabasePaths(%q, %q) = %v, want sidecar hard-link error", runtimePath, businessPath, err) + } +} + +func TestValidateDatabasePathsRejectsRuntimeStateHardLinkAlias(t *testing.T) { + directory := t.TempDir() + runtimePath := filepath.Join(directory, "runtime.db") + statePath := derivedRuntimeStatePath(runtimePath) + if err := os.WriteFile(runtimePath, []byte("runtime"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.Link(runtimePath, statePath); err != nil { + if errors.Is(err, syscall.EOPNOTSUPP) || errors.Is(err, syscall.EXDEV) || errors.Is(err, syscall.EPERM) { + t.Fatalf("hard links are required for this deterministic test: %v", err) + } + t.Fatal(err) + } + if err := validateDatabasePaths(runtimePath, filepath.Join(directory, "business.db")); err == nil || !strings.Contains(err.Error(), "must not alias") { + t.Fatalf("validateDatabasePaths(%q, %q) = %v, want runtime/state hard-link error", runtimePath, statePath, err) + } +} + +func TestValidateDatabasePathsRejectsParentTraversalAndSymlinkAliases(t *testing.T) { + directory := t.TempDir() + runtimePath := filepath.Join(directory, "runtime.db") + businessPath := filepath.Join(directory, "business.db") + if err := os.Symlink(filepath.Join(directory, "target.db"), filepath.Join(directory, "link-one.db")); err != nil { + t.Fatalf("create first symlink: %v", err) + } + if err := os.Symlink(filepath.Join(directory, "link-one.db"), filepath.Join(directory, "link-two.db")); err != nil { + t.Fatalf("create second symlink: %v", err) + } + if err := validateDatabasePaths(filepath.Join(directory, "link-two.db"), businessPath); err == nil || !strings.Contains(err.Error(), "symlink") { + t.Fatalf("multi-hop symlink validation = %v, want symlink error", err) + } + + traversalPath := filepath.Join(directory, "link-one.db") + string(os.PathSeparator) + ".." + string(os.PathSeparator) + "runtime.db" + if err := validateDatabasePaths(traversalPath, businessPath); err == nil || !strings.Contains(err.Error(), "must not contain '..'") { + t.Fatalf("symlink-aware parent traversal validation = %v, want parent traversal error", err) + } + + traversalRuntime := filepath.Join(directory, "runtime-alias.db") + if err := os.Symlink(runtimePath, traversalRuntime); err != nil { + t.Fatalf("create runtime symlink: %v", err) + } + if err := validateDatabasePaths(traversalRuntime, businessPath); err == nil || !strings.Contains(err.Error(), "symlink") { + t.Fatalf("runtime self-alias validation = %v, want symlink error", err) + } +} diff --git a/examples/shadow/conformance_core_test.go b/examples/shadow/conformance_core_test.go new file mode 100644 index 0000000..2756740 --- /dev/null +++ b/examples/shadow/conformance_core_test.go @@ -0,0 +1,190 @@ +package shadow_test + +import ( + "bytes" + "context" + "encoding/json" + "testing" + "testing/synctest" + "time" + + "github.com/suraciii/gor" + "github.com/suraciii/gor/clock" + "github.com/suraciii/gor/examples/shadow/domain" + "github.com/suraciii/gor/store" +) + +func TestConformance_TypedGrainsStatePresenceAndClear(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(1000, 0).UTC() + sourceClock := clock.NewFake(start) + stateStore := store.NewMemory() + reminderStore := store.NewMemory() + application := domain.NewMemoryApplicationStore() + rt := newConformanceRuntime(t, sourceClock, stateStore, reminderStore, application, nil, nil) + ctx := context.Background() + + device := gor.Ref[domain.Device](rt, "device-1") + workshop := gor.Ref[domain.Workshop](rt, "assembly") + if err := device.Report(ctx, "assembly", "temperature=20"); err != nil { + t.Fatalf("Report: %v", err) + } + if count, err := workshop.OnlineCount(ctx); err != nil || count != 1 { + t.Fatalf("typed cross-Grain OnlineCount = (%d, %v), want (1, nil)", count, err) + } + if exists, err := device.ShadowExists(ctx); err != nil || !exists { + t.Fatalf("ShadowExists after Report = (%v, %v), want (true, nil)", exists, err) + } + + if err := device.ClearShadow(ctx); err != nil { + t.Fatalf("ClearShadow: %v", err) + } + if exists, err := device.ShadowExists(ctx); err != nil || exists { + t.Fatalf("ShadowExists after ClearShadow = (%v, %v), want (false, nil)", exists, err) + } + if err := device.Configure(ctx, ""); err != nil { + t.Fatalf("Configure empty shadow: %v", err) + } + zero, err := device.Shadow(ctx) + if err != nil { + t.Fatalf("Shadow present zero: %v", err) + } + if zero != (domain.Shadow{}) { + t.Fatalf("present zero Shadow = %#v, want zero value", zero) + } + if exists, err := device.ShadowExists(ctx); err != nil || !exists { + t.Fatalf("ShadowExists for present zero = (%v, %v), want (true, nil)", exists, err) + } + if err := device.ClearShadow(ctx); err != nil { + t.Fatalf("second ClearShadow: %v", err) + } + + rt.Kill() + rt = newConformanceRuntime(t, sourceClock, stateStore, reminderStore, application, nil, nil) + defer rt.Kill() + if exists, err := gor.Ref[domain.Device](rt, "device-1").ShadowExists(ctx); err != nil || exists { + t.Fatalf("ShadowExists after reactivation = (%v, %v), want (false, nil)", exists, err) + } + }) +} + +func TestConformance_RequestContextIsCopiedAndNotPersisted(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(1100, 0).UTC() + sourceClock := clock.NewFake(start) + stateStore := store.NewMemory() + reminderStore := store.NewMemory() + application := domain.NewMemoryApplicationStore() + observed := make(chan domain.RecoveryObservation, 4) + rt := newConformanceRuntime(t, sourceClock, stateStore, reminderStore, application, observed, nil) + defer rt.Kill() + ctx, err := gor.WithRequestContext(context.Background(), "trace_id", "trace-1") + if err != nil { + t.Fatal(err) + } + + coordinator := gor.Ref[domain.RecoveryCoordinator](rt, domain.RecoveryCoordinatorKey) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + if err := gor.Ref[domain.Device](rt, "device-1").ReportAction(ctx, "action-1", "temperature=20"); err != nil { + t.Fatalf("ReportAction with Request Context: %v", err) + } + pending, err := application.ListPending(context.Background()) + if err != nil { + t.Fatalf("ListPending: %v", err) + } + if len(pending) != 1 || pending[0].TraceID != "trace-1" { + t.Fatalf("pending actions = %#v, want one copied trace ID", pending) + } + + stateRecord, err := stateStore.Read(context.Background(), store.GrainId{GrainType: gor.TypeName[domain.Device](), GrainKey: "device-1"}) + if err != nil { + t.Fatalf("read Device State: %v", err) + } + if bytes.Contains(stateRecord.Data, []byte("trace_id")) || bytes.Contains(stateRecord.Data, []byte("trace-1")) { + t.Fatalf("Device State contains Request Context: %s", stateRecord.Data) + } + rows, err := reminderStore.ListDue(context.Background(), start.Add(2*time.Second)) + if err != nil { + t.Fatalf("ListDue: %v", err) + } + if !hasReminder(rows, gor.TypeName[domain.RecoveryCoordinator](), domain.RecoveryCoordinatorKey, domain.RecoveryReminderName) { + t.Fatalf("Reminders = %#v, want fixed-key recovery Reminder", rows) + } + for _, row := range rows { + encoded, err := json.Marshal(row) + if err != nil { + t.Fatal(err) + } + if bytes.Contains(encoded, []byte("request_context")) || bytes.Contains(encoded, []byte("trace-1")) { + t.Fatalf("Reminder record contains Request Context: %s", encoded) + } + } + + sourceClock.Advance(domain.RecoveryInterval) + synctest.Wait() + observation := <-observed + if observation.TracePresent || observation.TraceID != nil { + t.Fatalf("Reminder Request Context = %#v, want absent", observation) + } + if !observation.Tick.FirstTickTime.Equal(start.Add(domain.RecoveryInterval)) || observation.Tick.Period != domain.RecoveryInterval { + t.Fatalf("Reminder TickStatus = %#v, want fixed first tick and period", observation.Tick) + } + }) +} + +func TestConformance_RestartRecoversPendingAction(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(1200, 0).UTC() + sourceClock := clock.NewFake(start) + stateStore := store.NewMemory() + reminderStore := store.NewMemory() + application := domain.NewMemoryApplicationStore() + observed := make(chan domain.RecoveryObservation, 4) + rt := newConformanceRuntime(t, sourceClock, stateStore, reminderStore, application, observed, nil) + coordinator := gor.Ref[domain.RecoveryCoordinator](rt, domain.RecoveryCoordinatorKey) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + ctx, err := gor.WithRequestContext(context.Background(), "trace_id", "trace-restart") + if err != nil { + t.Fatal(err) + } + if err := gor.Ref[domain.Device](rt, "device-1").ReportAction(ctx, "action-restart", "temperature=21"); err != nil { + t.Fatalf("ReportAction: %v", err) + } + rt.Kill() + + calls := make(chan gor.CallObservation, 16) + rt = newConformanceRuntime(t, sourceClock, stateStore, reminderStore, application, observed, calls) + defer rt.Kill() + if err := gor.Ref[domain.RecoveryCoordinator](rt, domain.RecoveryCoordinatorKey).Start(context.Background()); err != nil { + t.Fatalf("restart Start: %v", err) + } + sourceClock.Advance(domain.RecoveryInterval) + synctest.Wait() + observation := <-observed + if observation.TracePresent { + t.Fatal("recovery Reminder inherited Request Context") + } + record, applied, err := application.ReadApplied(context.Background(), "action-restart") + if err != nil { + t.Fatalf("ReadApplied: %v", err) + } + if !applied || record.TraceID != "trace-restart" || record.State != "temperature=21" { + t.Fatalf("applied record = (%#v, %v), want copied action receipt", record, applied) + } + pending, err := application.ListPending(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(pending) != 0 { + t.Fatalf("pending actions after recovery = %#v, want none", pending) + } + observations := drainCalls(calls) + if !containsCall(observations, "Recover", nil) || !containsCall(observations, "ApplyPending", nil) { + t.Fatalf("OnCall observations = %v, want Recover and typed ApplyPending", observations) + } + }) +} diff --git a/examples/shadow/conformance_failures_test.go b/examples/shadow/conformance_failures_test.go new file mode 100644 index 0000000..bcd4a4e --- /dev/null +++ b/examples/shadow/conformance_failures_test.go @@ -0,0 +1,230 @@ +package shadow_test + +import ( + "context" + "errors" + "testing" + "testing/synctest" + "time" + + "github.com/suraciii/gor" + "github.com/suraciii/gor/clock" + shadowdomain "github.com/suraciii/gor/examples/shadow/domain" + "github.com/suraciii/gor/store" +) + +func TestConformance_ReminderClaimHasOneWinner(t *testing.T) { + start := time.Unix(1300, 0).UTC() + reminderStore := store.NewMemory() + row := store.Reminder{ + GrainId: store.GrainId{GrainType: gor.TypeName[shadowdomain.RecoveryCoordinator](), GrainKey: shadowdomain.RecoveryCoordinatorKey}, + Name: shadowdomain.RecoveryReminderName, + Method: "Recover", + FirstTickTime: start, + DueAt: start, + Interval: shadowdomain.RecoveryInterval, + } + if err := reminderStore.Put(context.Background(), row); err != nil { + t.Fatal(err) + } + due, err := reminderStore.ListDue(context.Background(), start) + if err != nil || len(due) != 1 { + t.Fatalf("ListDue = (%#v, %v), want one row", due, err) + } + startClaims := make(chan struct{}) + results := make(chan claimResult, 2) + for range 2 { + go func() { + <-startClaims + won, err := reminderStore.Claim(context.Background(), due[0], start.Add(shadowdomain.RecoveryInterval)) + results <- claimResult{won: won, err: err} + }() + } + close(startClaims) + wins := 0 + for range 2 { + result := <-results + if result.err != nil { + t.Fatal(result.err) + } + if result.won { + wins++ + } + } + if wins != 1 { + t.Fatalf("Reminder claim winners = %d, want one", wins) + } +} + +func TestConformance_StopAfterClaimLeavesPendingForNextTick(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(1400, 0).UTC() + sourceClock := clock.NewFake(start) + stateStore := store.NewMemory() + baseReminders := store.NewMemory() + blocker := &blockingReminderStore{ + ReminderStore: baseReminders, + claimed: make(chan struct{}), + release: make(chan struct{}), + } + application := shadowdomain.NewMemoryApplicationStore() + rt := newConformanceRuntime(t, sourceClock, stateStore, blocker, application, nil, nil) + if err := gor.Ref[shadowdomain.RecoveryCoordinator](rt, shadowdomain.RecoveryCoordinatorKey).Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + if err := gor.Ref[shadowdomain.Device](rt, "device-1").ReportAction(context.Background(), "action-claim", "temperature=22"); err != nil { + t.Fatalf("ReportAction: %v", err) + } + sourceClock.Advance(shadowdomain.RecoveryInterval) + synctest.Wait() + <-blocker.claimed + + killDone := make(chan struct{}) + go func() { + rt.Kill() + close(killDone) + }() + <-rt.Done() + close(blocker.release) + <-killDone + if _, applied, err := application.ReadApplied(context.Background(), "action-claim"); err != nil || applied { + t.Fatalf("applied action after stop between claim and delivery = (%v, %v), want (nil, false)", err, applied) + } + + rt = newConformanceRuntime(t, sourceClock, stateStore, baseReminders, application, nil, nil) + defer rt.Kill() + if err := gor.Ref[shadowdomain.RecoveryCoordinator](rt, shadowdomain.RecoveryCoordinatorKey).Start(context.Background()); err != nil { + t.Fatalf("restart Start: %v", err) + } + sourceClock.Advance(shadowdomain.RecoveryInterval) + synctest.Wait() + if _, applied, err := application.ReadApplied(context.Background(), "action-claim"); err != nil || !applied { + t.Fatalf("applied action after next periodic tick = (%v, %v), want (nil, true)", err, applied) + } + }) +} + +func TestConformance_SafeRepeatAndUnknownResult(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(1500, 0).UTC() + sourceClock := clock.NewFake(start) + stateStore := store.NewMemory() + reminderStore := store.NewMemory() + baseApplication := shadowdomain.NewMemoryApplicationStore() + unknownErr := errors.New("application result is unknown") + application := &faultApplicationStore{ApplicationStore: baseApplication, afterCommit: unknownErr} + errorsSeen := make(chan gor.BackgroundError, 4) + calls := make(chan gor.CallObservation, 16) + rt := newConformanceRuntimeWithErrors(t, sourceClock, stateStore, reminderStore, application, nil, calls, errorsSeen) + defer rt.Kill() + if err := gor.Ref[shadowdomain.RecoveryCoordinator](rt, shadowdomain.RecoveryCoordinatorKey).Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + if err := gor.Ref[shadowdomain.Device](rt, "device-1").ReportAction(context.Background(), "action-unknown", "temperature=23"); err != nil { + t.Fatalf("ReportAction: %v", err) + } + sourceClock.Advance(shadowdomain.RecoveryInterval) + synctest.Wait() + background := <-errorsSeen + if !errors.Is(background.Err, unknownErr) { + t.Fatalf("OnError = %#v, want unknown result", background) + } + source, ok := background.Source.(gor.ReminderInvocation) + if !ok || source.Method != "Recover" { + t.Fatalf("OnError source = %#v, want ReminderInvocation{Recover}", background.Source) + } + if !hasCall(calls, "Recover", unknownErr) { + t.Fatalf("OnCall observations = %v, want Recover error", drainCalls(calls)) + } + record, applied, err := baseApplication.ReadApplied(context.Background(), "action-unknown") + if err != nil || !applied { + t.Fatalf("ReadApplied after unknown result = (%#v, %v, %v), want one receipt", record, applied, err) + } + if pending, err := baseApplication.ListPending(context.Background()); err != nil || len(pending) != 0 { + t.Fatalf("pending after unknown result = (%#v, %v), want none", pending, err) + } + if err := gor.Ref[shadowdomain.Device](rt, "device-1").ApplyPending(context.Background(), "action-unknown"); err != nil { + t.Fatalf("Safe Repeat ApplyPending: %v", err) + } + repeated, applied, err := baseApplication.ReadApplied(context.Background(), "action-unknown") + if err != nil || !applied || repeated != record { + t.Fatalf("receipt after Safe Repeat = (%#v, %v, %v), want unchanged receipt", repeated, applied, err) + } + }) +} + +func TestConformance_ReminderFailureRetriesPendingAndReportsError(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(1600, 0).UTC() + sourceClock := clock.NewFake(start) + stateStore := store.NewMemory() + reminderStore := store.NewMemory() + baseApplication := shadowdomain.NewMemoryApplicationStore() + beforeErr := errors.New("application commit failed before commit") + application := &faultApplicationStore{ApplicationStore: baseApplication, before: beforeErr} + errorsSeen := make(chan gor.BackgroundError, 4) + rt := newConformanceRuntimeWithErrors(t, sourceClock, stateStore, reminderStore, application, nil, nil, errorsSeen) + defer rt.Kill() + if err := gor.Ref[shadowdomain.RecoveryCoordinator](rt, shadowdomain.RecoveryCoordinatorKey).Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + if err := gor.Ref[shadowdomain.Device](rt, "device-1").ReportAction(context.Background(), "action-before", "temperature=24"); err != nil { + t.Fatalf("ReportAction: %v", err) + } + sourceClock.Advance(shadowdomain.RecoveryInterval) + synctest.Wait() + background := <-errorsSeen + if !errors.Is(background.Err, beforeErr) { + t.Fatalf("OnError = %#v, want pre-commit error", background) + } + if pending, err := baseApplication.ListPending(context.Background()); err != nil || len(pending) != 1 { + t.Fatalf("pending after failed Reminder = (%#v, %v), want one action", pending, err) + } + sourceClock.Advance(shadowdomain.RecoveryInterval) + synctest.Wait() + if _, applied, err := baseApplication.ReadApplied(context.Background(), "action-before"); err != nil || !applied { + t.Fatalf("applied after later Reminder = (%v, %v), want true", err, applied) + } + }) +} + +func TestConformance_CancelRecoveryReminderClearsScheduleAndState(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(1700, 0).UTC() + sourceClock := clock.NewFake(start) + stateStore := store.NewMemory() + reminderStore := store.NewMemory() + application := shadowdomain.NewMemoryApplicationStore() + observed := make(chan shadowdomain.RecoveryObservation, 4) + rt := newConformanceRuntime(t, sourceClock, stateStore, reminderStore, application, observed, nil) + defer rt.Kill() + coordinator := gor.Ref[shadowdomain.RecoveryCoordinator](rt, shadowdomain.RecoveryCoordinatorKey) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + if err := coordinator.Stop(context.Background()); err != nil { + t.Fatalf("Stop: %v", err) + } + rows, err := reminderStore.ListDue(context.Background(), start.Add(2*shadowdomain.RecoveryInterval)) + if err != nil { + t.Fatal(err) + } + if hasReminder(rows, gor.TypeName[shadowdomain.RecoveryCoordinator](), shadowdomain.RecoveryCoordinatorKey, shadowdomain.RecoveryReminderName) { + t.Fatalf("recovery Reminder remains after Stop: %#v", rows) + } + record, err := stateStore.Read(context.Background(), store.GrainId{GrainType: gor.TypeName[shadowdomain.RecoveryCoordinator](), GrainKey: shadowdomain.RecoveryCoordinatorKey}) + if err != nil { + t.Fatal(err) + } + if string(record.Data) != `{}` { + t.Fatalf("coordinator State after Stop = %s, want cleared record", record.Data) + } + sourceClock.Advance(2 * shadowdomain.RecoveryInterval) + synctest.Wait() + select { + case observation := <-observed: + t.Fatalf("recovery Call after cancellation: %#v", observation) + default: + } + }) +} diff --git a/examples/shadow/conformance_helpers_test.go b/examples/shadow/conformance_helpers_test.go new file mode 100644 index 0000000..0ecbada --- /dev/null +++ b/examples/shadow/conformance_helpers_test.go @@ -0,0 +1,135 @@ +package shadow_test + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/suraciii/gor" + "github.com/suraciii/gor/clock" + shadow "github.com/suraciii/gor/examples/shadow" + "github.com/suraciii/gor/examples/shadow/domain" + "github.com/suraciii/gor/store" +) + +type claimResult struct { + won bool + err error +} + +type blockingReminderStore struct { + store.ReminderStore + claimed chan struct{} + release chan struct{} +} + +func (s *blockingReminderStore) Claim(ctx context.Context, reminder store.Reminder, nextDueAt time.Time) (bool, error) { + won, err := s.ReminderStore.Claim(ctx, reminder, nextDueAt) + if err != nil || !won { + return won, err + } + close(s.claimed) + <-s.release + return won, nil +} + +type faultApplicationStore struct { + domain.ApplicationStore + before error + afterCommit error + beforeUsed atomic.Bool + afterUsed atomic.Bool +} + +func (s *faultApplicationStore) ApplyPending(ctx context.Context, actionID string) error { + if s.before != nil && s.beforeUsed.CompareAndSwap(false, true) { + return s.before + } + if err := s.ApplicationStore.ApplyPending(ctx, actionID); err != nil { + return err + } + if s.afterCommit != nil && s.afterUsed.CompareAndSwap(false, true) { + return s.afterCommit + } + return nil +} + +func newConformanceRuntime(t *testing.T, sourceClock *clock.Fake, stateStore store.Store, reminderStore store.ReminderStore, application domain.ApplicationStore, observed chan<- domain.RecoveryObservation, calls chan gor.CallObservation) *gor.Runtime { + t.Helper() + return newConformanceRuntimeWithErrors(t, sourceClock, stateStore, reminderStore, application, observed, calls, nil) +} + +func newConformanceRuntimeWithErrors(t *testing.T, sourceClock *clock.Fake, stateStore store.Store, reminderStore store.ReminderStore, application domain.ApplicationStore, observed chan<- domain.RecoveryObservation, calls chan gor.CallObservation, errorsSeen chan gor.BackgroundError) *gor.Runtime { + t.Helper() + if errorsSeen == nil { + errorsSeen = make(chan gor.BackgroundError, 16) + } + options := []gor.Option{ + gor.WithStore(stateStore), + gor.WithReminderStore(reminderStore), + gor.WithClock(sourceClock), + gor.WithIdleTimeout(0), + gor.WithEvictionInterval(0), + gor.WithReminderInterval(domain.RecoveryInterval), + gor.OnError(func(event gor.BackgroundError) { errorsSeen <- event }), + } + if calls != nil { + options = append(options, gor.OnCall(func(observation gor.CallObservation) { calls <- observation })) + } + rt, err := gor.New(options...) + if err != nil { + t.Fatal(err) + } + if observed == nil { + if err := shadow.RegisterConformance(rt, application); err != nil { + rt.Kill() + t.Fatal(err) + } + } else if err := shadow.RegisterConformanceWithObservation(rt, application, observed); err != nil { + rt.Kill() + t.Fatal(err) + } + return rt +} + +func hasReminder(rows []store.Reminder, grainType, grainKey, name string) bool { + for _, row := range rows { + if row.GrainId.GrainType == grainType && row.GrainId.GrainKey == grainKey && row.Name == name { + return true + } + } + return false +} + +func hasCall(calls chan gor.CallObservation, method string, wantErr error) bool { + found := false + for len(calls) > 0 { + observation := <-calls + if observation.Method == method && (wantErr == nil || errors.Is(observation.Err, wantErr)) { + found = true + } + } + return found +} + +func drainCalls(calls chan gor.CallObservation) []gor.CallObservation { + var result []gor.CallObservation + for len(calls) > 0 { + result = append(result, <-calls) + } + return result +} + +func containsCall(observations []gor.CallObservation, method string, wantErr error) bool { + for _, observation := range observations { + if observation.Method == method && (wantErr == nil || errors.Is(observation.Err, wantErr)) { + return true + } + } + return false +} + +var _ domain.ApplicationStore = (*faultApplicationStore)(nil) +var _ store.ReminderStore = (*blockingReminderStore)(nil) diff --git a/examples/shadow/conformance_report_action_test.go b/examples/shadow/conformance_report_action_test.go new file mode 100644 index 0000000..cad6b24 --- /dev/null +++ b/examples/shadow/conformance_report_action_test.go @@ -0,0 +1,52 @@ +package shadow_test + +import ( + "context" + "errors" + "testing" + "testing/synctest" + "time" + + "github.com/suraciii/gor" + "github.com/suraciii/gor/clock" + "github.com/suraciii/gor/examples/shadow/domain" + "github.com/suraciii/gor/store" +) + +func TestConformance_ReportActionConflictLeavesShadowUnchanged(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + sourceClock := clock.NewFake(time.Unix(1900, 0).UTC()) + application := domain.NewMemoryApplicationStore() + rt := newConformanceRuntime(t, sourceClock, store.NewMemory(), store.NewMemory(), application, nil, nil) + defer rt.Kill() + + device := gor.Ref[domain.Device](rt, "device-1") + if err := device.ReportAction(context.Background(), "action-existing", "temperature=20"); err != nil { + t.Fatalf("seed ReportAction: %v", err) + } + before, err := device.Shadow(context.Background()) + if err != nil { + t.Fatalf("read shadow before conflict: %v", err) + } + if err := application.SavePending(context.Background(), domain.PendingAction{ + ActionID: "action-conflict", + DeviceKey: "device-1", + State: "temperature=21", + TraceID: "trace-existing", + }); err != nil { + t.Fatalf("seed conflicting action: %v", err) + } + + err = device.ReportAction(context.Background(), "action-conflict", "temperature=22") + if !errors.Is(err, domain.ErrPendingActionConflict) { + t.Fatalf("conflicting ReportAction error = %v, want ErrPendingActionConflict", err) + } + after, err := device.Shadow(context.Background()) + if err != nil { + t.Fatalf("read shadow after conflict: %v", err) + } + if after != before { + t.Fatalf("shadow after deterministic conflict = %#v, want unchanged %#v", after, before) + } + }) +} diff --git a/examples/shadow/conformance_stop_test.go b/examples/shadow/conformance_stop_test.go new file mode 100644 index 0000000..302a9b6 --- /dev/null +++ b/examples/shadow/conformance_stop_test.go @@ -0,0 +1,134 @@ +package shadow_test + +import ( + "context" + "errors" + "testing" + "testing/synctest" + "time" + + "github.com/suraciii/gor" + "github.com/suraciii/gor/clock" + "github.com/suraciii/gor/examples/shadow/domain" + "github.com/suraciii/gor/store" +) + +func TestConformance_StopInterruptionRestartsCoordinator(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(1800, 0).UTC() + sourceClock := clock.NewFake(start) + stateStore := store.NewMemory() + baseReminders := store.NewMemory() + blockingReminders := &blockingDeleteReminderStore{ + ReminderStore: baseReminders, + deleted: make(chan struct{}), + release: make(chan struct{}), + } + application := domain.NewMemoryApplicationStore() + observed := make(chan domain.RecoveryObservation, 4) + rt := newConformanceRuntime(t, sourceClock, stateStore, blockingReminders, application, observed, nil) + coordinator := gor.Ref[domain.RecoveryCoordinator](rt, domain.RecoveryCoordinatorKey) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + before, err := baseReminders.ListDue(context.Background(), start.Add(2*domain.RecoveryInterval)) + if err != nil { + t.Fatal(err) + } + beforeReminder := findRecoveryReminder(t, before) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatalf("idempotent Start: %v", err) + } + after, err := baseReminders.ListDue(context.Background(), start.Add(2*domain.RecoveryInterval)) + if err != nil { + t.Fatal(err) + } + afterReminder := findRecoveryReminder(t, after) + if !afterReminder.FirstTickTime.Equal(beforeReminder.FirstTickTime) { + t.Fatalf("idempotent Start changed FirstTickTime from %s to %s", beforeReminder.FirstTickTime, afterReminder.FirstTickTime) + } + if err := gor.Ref[domain.Device](rt, "device-1").ReportAction(context.Background(), "action-stop", "temperature=25"); err != nil { + t.Fatalf("ReportAction: %v", err) + } + + stopDone := make(chan error, 1) + go func() { + stopDone <- coordinator.Stop(context.Background()) + }() + <-blockingReminders.deleted + + coordinatorRecord, err := stateStore.Read(context.Background(), store.GrainId{ + GrainType: gor.TypeName[domain.RecoveryCoordinator](), + GrainKey: domain.RecoveryCoordinatorKey, + }) + if err != nil { + t.Fatal(err) + } + if string(coordinatorRecord.Data) != `{}` { + t.Fatalf("Coordinator State while Cancel is blocked = %s, want cleared state", coordinatorRecord.Data) + } + remaining, err := baseReminders.ListDue(context.Background(), start.Add(2*domain.RecoveryInterval)) + if err != nil { + t.Fatal(err) + } + if hasReminder(remaining, gor.TypeName[domain.RecoveryCoordinator](), domain.RecoveryCoordinatorKey, domain.RecoveryReminderName) { + t.Fatalf("recovery Reminder remains after Delete completed: %#v", remaining) + } + + killDone := make(chan struct{}) + go func() { + rt.Kill() + close(killDone) + }() + <-rt.Done() + close(blockingReminders.release) + <-killDone + if err := <-stopDone; err != nil && !errors.Is(err, context.Canceled) { + t.Fatalf("Stop after interrupted process: %v", err) + } + + rt = newConformanceRuntime(t, sourceClock, stateStore, baseReminders, application, observed, nil) + defer rt.Kill() + if err := gor.Ref[domain.RecoveryCoordinator](rt, domain.RecoveryCoordinatorKey).Start(context.Background()); err != nil { + t.Fatalf("restart Start: %v", err) + } + rows, err := baseReminders.ListDue(context.Background(), start.Add(2*domain.RecoveryInterval)) + if err != nil { + t.Fatal(err) + } + findRecoveryReminder(t, rows) + sourceClock.Advance(domain.RecoveryInterval) + synctest.Wait() + if _, applied, err := application.ReadApplied(context.Background(), "action-stop"); err != nil || !applied { + t.Fatalf("action after restart recovery = (%v, %v), want (nil, true)", err, applied) + } + }) +} + +type blockingDeleteReminderStore struct { + store.ReminderStore + deleted chan struct{} + release chan struct{} +} + +func (s *blockingDeleteReminderStore) Delete(ctx context.Context, id store.GrainId, name string) error { + if err := s.ReminderStore.Delete(ctx, id, name); err != nil { + return err + } + close(s.deleted) + <-s.release + return nil +} + +func findRecoveryReminder(t *testing.T, rows []store.Reminder) store.Reminder { + t.Helper() + for _, row := range rows { + if row.GrainId.GrainType == gor.TypeName[domain.RecoveryCoordinator]() && row.GrainId.GrainKey == domain.RecoveryCoordinatorKey && row.Name == domain.RecoveryReminderName { + return row + } + } + t.Fatalf("recovery Reminder missing from %#v", rows) + return store.Reminder{} +} + +var _ store.ReminderStore = (*blockingDeleteReminderStore)(nil) diff --git a/examples/shadow/domain/application.go b/examples/shadow/domain/application.go new file mode 100644 index 0000000..7b2189e --- /dev/null +++ b/examples/shadow/domain/application.go @@ -0,0 +1,335 @@ +package domain + +import ( + "context" + "database/sql" + "errors" + "fmt" + "net/url" + "path/filepath" + "sort" + "sync" + + _ "modernc.org/sqlite" +) + +// ApplicationStore owns business records for the conformance example. It is +// separate from the stores used by the Grain Runtime. +type ApplicationStore interface { + SavePending(context.Context, PendingAction) error + ListPending(context.Context) ([]PendingAction, error) + ApplyPending(context.Context, string) error + ReadApplied(context.Context, string) (AppliedRecord, bool, error) + Close() error +} + +// PendingAction is an application-owned Business Action waiting for recovery. +type PendingAction struct { + ActionID string + DeviceKey string + State string + TraceID string +} + +// AppliedRecord is the application receipt for one ActionID. +type AppliedRecord struct { + ActionID string + DeviceKey string + State string + TraceID string +} + +var ( + // ErrPendingActionConflict reports reuse of an ActionID with a new payload. + ErrPendingActionConflict = errors.New("application action ID has a different payload") + // ErrPendingActionNotFound reports an action that is neither pending nor applied. + ErrPendingActionNotFound = errors.New("application pending action was not found") +) + +// MemoryApplicationStore is an in-memory ApplicationStore for deterministic +// example tests. +type MemoryApplicationStore struct { + mu sync.Mutex + pending map[string]PendingAction + applied map[string]AppliedRecord +} + +var _ ApplicationStore = (*MemoryApplicationStore)(nil) + +// NewMemoryApplicationStore returns an empty ApplicationStore. +func NewMemoryApplicationStore() *MemoryApplicationStore { + return &MemoryApplicationStore{ + pending: make(map[string]PendingAction), + applied: make(map[string]AppliedRecord), + } +} + +// SavePending inserts an action, or accepts an identical repeat. +func (s *MemoryApplicationStore) SavePending(ctx context.Context, action PendingAction) error { + if err := ctx.Err(); err != nil { + return err + } + if action.ActionID == "" { + return errors.New("application action ID is empty") + } + s.mu.Lock() + defer s.mu.Unlock() + if current, ok := s.pending[action.ActionID]; ok { + if current != action { + return ErrPendingActionConflict + } + return nil + } + if current, ok := s.applied[action.ActionID]; ok { + if current != (AppliedRecord(action)) { + return ErrPendingActionConflict + } + return nil + } + s.pending[action.ActionID] = action + return nil +} + +// ListPending returns pending actions in ActionID order. +func (s *MemoryApplicationStore) ListPending(ctx context.Context) ([]PendingAction, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + s.mu.Lock() + defer s.mu.Unlock() + result := make([]PendingAction, 0, len(s.pending)) + for _, action := range s.pending { + result = append(result, action) + } + sort.Slice(result, func(i, j int) bool { return result[i].ActionID < result[j].ActionID }) + return result, nil +} + +// ApplyPending applies one action and creates one receipt. Repeating an +// applied ActionID succeeds without changing the receipt. +func (s *MemoryApplicationStore) ApplyPending(ctx context.Context, actionID string) error { + if err := ctx.Err(); err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + if _, ok := s.applied[actionID]; ok { + return nil + } + action, ok := s.pending[actionID] + if !ok { + return ErrPendingActionNotFound + } + s.applied[actionID] = AppliedRecord(action) + delete(s.pending, actionID) + return nil +} + +// ReadApplied reads one application receipt. +func (s *MemoryApplicationStore) ReadApplied(ctx context.Context, actionID string) (AppliedRecord, bool, error) { + if err := ctx.Err(); err != nil { + return AppliedRecord{}, false, err + } + s.mu.Lock() + defer s.mu.Unlock() + record, ok := s.applied[actionID] + return record, ok, nil +} + +// Close releases no resources and is provided to keep the store boundary equal +// to the durable implementation. +func (s *MemoryApplicationStore) Close() error { + return nil +} + +// SQLiteApplicationStore is the durable ApplicationStore for the example. It +// owns only the pending_actions and applied_records tables in its own file. +type SQLiteApplicationStore struct { + db *sql.DB +} + +var _ ApplicationStore = (*SQLiteApplicationStore)(nil) + +// OpenSQLiteApplicationStore opens or creates a business database at path. +func OpenSQLiteApplicationStore(path string) (*SQLiteApplicationStore, error) { + db, err := sql.Open("sqlite", applicationSQLiteDSN(path)) + if err != nil { + return nil, err + } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + if err := db.Ping(); err != nil { + db.Close() + return nil, fmt.Errorf("open application database %q: %w", path, err) + } + if _, err := db.Exec(` +CREATE TABLE IF NOT EXISTS pending_actions ( + action_id TEXT PRIMARY KEY, + device_key TEXT NOT NULL, + state TEXT NOT NULL, + trace_id TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS applied_records ( + action_id TEXT PRIMARY KEY, + device_key TEXT NOT NULL, + state TEXT NOT NULL, + trace_id TEXT NOT NULL +)`); err != nil { + db.Close() + return nil, fmt.Errorf("create application schema: %w", err) + } + return &SQLiteApplicationStore{db: db}, nil +} + +func applicationSQLiteDSN(path string) string { + absolute, _ := filepath.Abs(path) + fileURI := (&url.URL{Scheme: "file", Path: filepath.ToSlash(absolute)}).String() + return fmt.Sprintf("%s?_pragma=journal_mode(WAL)&_pragma=synchronous(FULL)&_pragma=busy_timeout(5000)", fileURI) +} + +// SavePending inserts an action, or accepts an identical repeat. +func (s *SQLiteApplicationStore) SavePending(ctx context.Context, action PendingAction) error { + if err := ctx.Err(); err != nil { + return err + } + if action.ActionID == "" { + return errors.New("application action ID is empty") + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + + current, found, err := readPendingTx(ctx, tx, action.ActionID) + if err != nil { + return err + } + if found { + if current != action { + return ErrPendingActionConflict + } + return tx.Commit() + } + applied, found, err := readAppliedTx(ctx, tx, action.ActionID) + if err != nil { + return err + } + if found { + if applied != AppliedRecord(action) { + return ErrPendingActionConflict + } + return tx.Commit() + } + if _, err := tx.ExecContext(ctx, + `INSERT INTO pending_actions (action_id, device_key, state, trace_id) VALUES (?, ?, ?, ?)`, + action.ActionID, action.DeviceKey, action.State, action.TraceID, + ); err != nil { + return err + } + return tx.Commit() +} + +// ListPending returns pending actions in ActionID order. +func (s *SQLiteApplicationStore) ListPending(ctx context.Context) ([]PendingAction, error) { + rows, err := s.db.QueryContext(ctx, ` +SELECT action_id, device_key, state, trace_id +FROM pending_actions +ORDER BY action_id`) + if err != nil { + return nil, err + } + defer rows.Close() + var result []PendingAction + for rows.Next() { + var action PendingAction + if err := rows.Scan(&action.ActionID, &action.DeviceKey, &action.State, &action.TraceID); err != nil { + return nil, err + } + result = append(result, action) + } + if err := rows.Err(); err != nil { + return nil, err + } + return result, nil +} + +// ApplyPending applies one action in one application transaction. The unique +// ActionID makes a repeat a successful no-op. +func (s *SQLiteApplicationStore) ApplyPending(ctx context.Context, actionID string) error { + if err := ctx.Err(); err != nil { + return err + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + if _, found, err := readAppliedTx(ctx, tx, actionID); err != nil { + return err + } else if found { + return tx.Commit() + } + action, found, err := readPendingTx(ctx, tx, actionID) + if err != nil { + return err + } + if !found { + return ErrPendingActionNotFound + } + if _, err := tx.ExecContext(ctx, ` +INSERT INTO applied_records (action_id, device_key, state, trace_id) +VALUES (?, ?, ?, ?)`, action.ActionID, action.DeviceKey, action.State, action.TraceID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM pending_actions WHERE action_id = ?`, actionID); err != nil { + return err + } + return tx.Commit() +} + +// ReadApplied reads one application receipt. +func (s *SQLiteApplicationStore) ReadApplied(ctx context.Context, actionID string) (AppliedRecord, bool, error) { + return readApplied(ctx, s.db, actionID) +} + +// Close closes the business database. +func (s *SQLiteApplicationStore) Close() error { + return s.db.Close() +} + +type applicationQuery interface { + QueryRowContext(context.Context, string, ...any) *sql.Row +} + +func readPending(ctx context.Context, db applicationQuery, actionID string) (PendingAction, bool, error) { + var action PendingAction + err := db.QueryRowContext(ctx, ` +SELECT action_id, device_key, state, trace_id +FROM pending_actions WHERE action_id = ?`, actionID). + Scan(&action.ActionID, &action.DeviceKey, &action.State, &action.TraceID) + if errors.Is(err, sql.ErrNoRows) { + return PendingAction{}, false, nil + } + return action, err == nil, err +} + +func readApplied(ctx context.Context, db applicationQuery, actionID string) (AppliedRecord, bool, error) { + var record AppliedRecord + err := db.QueryRowContext(ctx, ` +SELECT action_id, device_key, state, trace_id +FROM applied_records WHERE action_id = ?`, actionID). + Scan(&record.ActionID, &record.DeviceKey, &record.State, &record.TraceID) + if errors.Is(err, sql.ErrNoRows) { + return AppliedRecord{}, false, nil + } + return record, err == nil, err +} + +func readPendingTx(ctx context.Context, tx *sql.Tx, actionID string) (PendingAction, bool, error) { + return readPending(ctx, tx, actionID) +} + +func readAppliedTx(ctx context.Context, tx *sql.Tx, actionID string) (AppliedRecord, bool, error) { + return readApplied(ctx, tx, actionID) +} diff --git a/examples/shadow/domain/application_test.go b/examples/shadow/domain/application_test.go new file mode 100644 index 0000000..42c7a11 --- /dev/null +++ b/examples/shadow/domain/application_test.go @@ -0,0 +1,118 @@ +package domain + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" +) + +func TestApplicationStore_SafeRepeatAndActionIDConflict(t *testing.T) { + stores := []ApplicationStore{NewMemoryApplicationStore()} + business, err := OpenSQLiteApplicationStore(filepath.Join(t.TempDir(), "business.db")) + if err != nil { + t.Fatal(err) + } + stores = append(stores, business) + for _, application := range stores { + t.Run(storeName(application), func(t *testing.T) { + t.Cleanup(func() { _ = application.Close() }) + ctx := context.Background() + first := PendingAction{ActionID: "b", DeviceKey: "device-1", State: "two", TraceID: "trace-b"} + second := PendingAction{ActionID: "a", DeviceKey: "device-1", State: "one", TraceID: "trace-a"} + if err := application.SavePending(ctx, first); err != nil { + t.Fatal(err) + } + if err := application.SavePending(ctx, first); err != nil { + t.Fatalf("identical SavePending: %v", err) + } + if err := application.SavePending(ctx, second); err != nil { + t.Fatal(err) + } + if err := application.SavePending(ctx, PendingAction{ActionID: "a", DeviceKey: "device-1", State: "changed"}); !errors.Is(err, ErrPendingActionConflict) { + t.Fatalf("conflicting SavePending = %v, want conflict", err) + } + pending, err := application.ListPending(ctx) + if err != nil || len(pending) != 2 || pending[0].ActionID != "a" || pending[1].ActionID != "b" { + t.Fatalf("ListPending = (%#v, %v), want ActionID order", pending, err) + } + if err := application.ApplyPending(ctx, "a"); err != nil { + t.Fatal(err) + } + if err := application.ApplyPending(ctx, "a"); err != nil { + t.Fatalf("Safe Repeat ApplyPending: %v", err) + } + record, ok, err := application.ReadApplied(ctx, "a") + if err != nil || !ok || record.State != "one" || record.TraceID != "trace-a" { + t.Fatalf("ReadApplied = (%#v, %v, %v), want one receipt", record, ok, err) + } + }) + } +} + +func TestSQLiteApplicationStore_PersistsPendingAndAppliedRecords(t *testing.T) { + path := filepath.Join(t.TempDir(), "business.db") + first, err := OpenSQLiteApplicationStore(path) + if err != nil { + t.Fatal(err) + } + if err := first.SavePending(context.Background(), PendingAction{ActionID: "persisted", DeviceKey: "device-1", State: "value", TraceID: "trace"}); err != nil { + first.Close() + t.Fatal(err) + } + if err := first.Close(); err != nil { + t.Fatal(err) + } + second, err := OpenSQLiteApplicationStore(path) + if err != nil { + t.Fatal(err) + } + if err := second.ApplyPending(context.Background(), "persisted"); err != nil { + second.Close() + t.Fatal(err) + } + defer second.Close() + record, ok, err := second.ReadApplied(context.Background(), "persisted") + if err != nil || !ok || record.ActionID != "persisted" { + t.Fatalf("ReadApplied after reopen = (%#v, %v, %v), want persisted receipt", record, ok, err) + } +} + +func TestSQLiteApplicationStore_PathWithURICharactersUsesRequestedFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "business?#.db") + first, err := OpenSQLiteApplicationStore(path) + if err != nil { + t.Fatal(err) + } + if err := first.SavePending(context.Background(), PendingAction{ActionID: "uri", DeviceKey: "device-1", State: "value", TraceID: "trace"}); err != nil { + first.Close() + t.Fatal(err) + } + if err := first.Close(); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(path); err != nil { + t.Fatalf("requested Application database %q: %v", path, err) + } + + second, err := OpenSQLiteApplicationStore(path) + if err != nil { + t.Fatal(err) + } + defer second.Close() + pending, err := second.ListPending(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(pending) != 1 || pending[0].ActionID != "uri" { + t.Fatalf("pending after reopen = %#v, want URI-path record", pending) + } +} + +func storeName(application ApplicationStore) string { + if _, ok := application.(*MemoryApplicationStore); ok { + return "memory" + } + return "sqlite" +} diff --git a/examples/shadow/domain/domain.go b/examples/shadow/domain/domain.go index fe4be3d..08b1879 100644 --- a/examples/shadow/domain/domain.go +++ b/examples/shadow/domain/domain.go @@ -2,6 +2,8 @@ package domain import ( "context" + "errors" + "fmt" "log" "time" @@ -34,11 +36,38 @@ type Shadow struct { //gor:grain type Device interface { Report(ctx context.Context, workshopID string, state string) error + ReportAction(ctx context.Context, actionID string, state string) error Configure(ctx context.Context, configuration string) error Shadow(ctx context.Context) (Shadow, error) + ShadowExists(ctx context.Context) (bool, error) + ClearShadow(ctx context.Context) error + ApplyPending(ctx context.Context, actionID string) error MarkOffline(ctx context.Context, tick gor.TickStatus) error } +//gor:grain +type RecoveryCoordinator interface { + Start(ctx context.Context) error + Stop(ctx context.Context) error + Recover(ctx context.Context, tick gor.TickStatus) error +} + +const ( + RecoveryCoordinatorKey = "recovery" + RecoveryReminderName = "recovery" + RecoveryInterval = time.Second +) + +type CoordinatorState struct { + Running bool +} + +type RecoveryObservation struct { + Tick gor.TickStatus + TraceID any + TracePresent bool +} + //gor:grain type Workshop interface { DeviceOnline(ctx context.Context, deviceID string) error @@ -51,23 +80,33 @@ type device struct { id gor.GrainId shadow gor.State[Shadow] schedule gor.Reminder[Device] + application ApplicationStore lifecycleEvents chan<- LifecycleEvent } func NewDevice(b *gor.Binder) Device { - return newDevice(b, nil) + return newDevice(b, nil, nil) } func NewDeviceWithLifecycle(b *gor.Binder, events chan<- LifecycleEvent) Device { - return newDevice(b, events) + return newDevice(b, events, nil) +} + +func NewDeviceWithApplication(b *gor.Binder, application ApplicationStore) Device { + return newDevice(b, nil, application) +} + +func NewDeviceWithApplicationAndLifecycle(b *gor.Binder, application ApplicationStore, events chan<- LifecycleEvent) Device { + return newDevice(b, events, application) } -func newDevice(b *gor.Binder, events chan<- LifecycleEvent) Device { +func newDevice(b *gor.Binder, events chan<- LifecycleEvent, application ApplicationStore) Device { return &device{ binder: b, id: gor.Self(b), shadow: gor.NewState[Shadow](b, "shadow"), schedule: gor.NewReminder[Device](b), + application: application, lifecycleEvents: events, } } @@ -102,6 +141,32 @@ func (d *device) Report(ctx context.Context, workshopID string, state string) er return nil } +func (d *device) ReportAction(ctx context.Context, actionID string, state string) error { + if d.application == nil { + return errors.New("application store is not configured") + } + if actionID == "" { + return errors.New("application action ID is empty") + } + traceID, err := traceIDFromContext(ctx) + if err != nil { + return err + } + if err := d.application.SavePending(ctx, PendingAction{ + ActionID: actionID, + DeviceKey: d.id.GrainKey, + State: state, + TraceID: traceID, + }); err != nil { + return err + } + shadow := d.shadow.Get() + shadow.ReportedState = state + shadow.ReportedAt = gor.Now(d.binder) + shadow.Online = true + return d.shadow.Set(ctx, shadow) +} + func (d *device) Configure(ctx context.Context, configuration string) error { shadow := d.shadow.Get() shadow.Configuration = configuration @@ -112,6 +177,21 @@ func (d *device) Shadow(context.Context) (Shadow, error) { return d.shadow.Get(), nil } +func (d *device) ShadowExists(context.Context) (bool, error) { + return d.shadow.Exists(), nil +} + +func (d *device) ClearShadow(ctx context.Context) error { + return d.shadow.Clear(ctx) +} + +func (d *device) ApplyPending(ctx context.Context, actionID string) error { + if d.application == nil { + return errors.New("application store is not configured") + } + return d.application.ApplyPending(ctx, actionID) +} + func (d *device) OnActivate(context.Context) error { log.Printf("%s activated", d.id.GrainKey) d.emitLifecycle(LifecycleActivated) @@ -130,6 +210,18 @@ func (d *device) emitLifecycle(kind string) { } } +func traceIDFromContext(ctx context.Context) (string, error) { + value, ok := gor.RequestContextValue(ctx, "trace_id") + if !ok { + return "", nil + } + traceID, ok := value.(string) + if !ok { + return "", fmt.Errorf("trace_id has type %T, want string", value) + } + return traceID, nil +} + func (d *device) MarkOffline(ctx context.Context, _ gor.TickStatus) error { shadow := d.shadow.Get() shadow.Online = false diff --git a/examples/shadow/domain/gorgen/generated.go b/examples/shadow/domain/gorgen/generated.go index b1bb216..9528eff 100644 --- a/examples/shadow/domain/gorgen/generated.go +++ b/examples/shadow/domain/gorgen/generated.go @@ -13,6 +13,26 @@ type deviceProxy struct { rt gor.Invoker } +type deviceApplyPendingRequest struct { + A0 string +} +type deviceApplyPendingReply struct{} + +func (p *deviceProxy) ApplyPending(ctx context.Context, actionID string) error { + var reply deviceApplyPendingReply + err := p.rt.Invoke(ctx, p.id, "ApplyPending", &deviceApplyPendingRequest{A0: actionID}, &reply) + return err +} + +type deviceClearShadowRequest struct{} +type deviceClearShadowReply struct{} + +func (p *deviceProxy) ClearShadow(ctx context.Context) error { + var reply deviceClearShadowReply + err := p.rt.Invoke(ctx, p.id, "ClearShadow", &deviceClearShadowRequest{}, &reply) + return err +} + type deviceConfigureRequest struct { A0 string } @@ -47,6 +67,18 @@ func (p *deviceProxy) Report(ctx context.Context, workshopID string, state strin return err } +type deviceReportActionRequest struct { + A0 string + A1 string +} +type deviceReportActionReply struct{} + +func (p *deviceProxy) ReportAction(ctx context.Context, actionID string, state string) error { + var reply deviceReportActionReply + err := p.rt.Invoke(ctx, p.id, "ReportAction", &deviceReportActionRequest{A0: actionID, A1: state}, &reply) + return err +} + type deviceShadowRequest struct{} type deviceShadowReply struct { R0 domain.Shadow @@ -58,8 +90,26 @@ func (p *deviceProxy) Shadow(ctx context.Context) (domain.Shadow, error) { return reply.R0, err } +type deviceShadowExistsRequest struct{} +type deviceShadowExistsReply struct { + R0 bool +} + +func (p *deviceProxy) ShadowExists(ctx context.Context) (bool, error) { + var reply deviceShadowExistsReply + err := p.rt.Invoke(ctx, p.id, "ShadowExists", &deviceShadowExistsRequest{}, &reply) + return reply.R0, err +} + func dispatchDevice(ctx context.Context, instance domain.Device, method string, args any, reply any) error { switch method { + case "ApplyPending": + typedArgs := args.(*deviceApplyPendingRequest) + err := instance.ApplyPending(ctx, typedArgs.A0) + return err + case "ClearShadow": + err := instance.ClearShadow(ctx) + return err case "Configure": typedArgs := args.(*deviceConfigureRequest) err := instance.Configure(ctx, typedArgs.A0) @@ -72,11 +122,20 @@ func dispatchDevice(ctx context.Context, instance domain.Device, method string, typedArgs := args.(*deviceReportRequest) err := instance.Report(ctx, typedArgs.A0, typedArgs.A1) return err + case "ReportAction": + typedArgs := args.(*deviceReportActionRequest) + err := instance.ReportAction(ctx, typedArgs.A0, typedArgs.A1) + return err case "Shadow": typedReply := reply.(*deviceShadowReply) r0, err := instance.Shadow(ctx) typedReply.R0 = r0 return err + case "ShadowExists": + typedReply := reply.(*deviceShadowExistsReply) + r0, err := instance.ShadowExists(ctx) + typedReply.R0 = r0 + return err default: return fmt.Errorf("unknown method %q", method) } @@ -84,14 +143,22 @@ func dispatchDevice(ctx context.Context, instance domain.Device, method string, func newDeviceCall(method string) (args any, reply any) { switch method { + case "ApplyPending": + return &deviceApplyPendingRequest{}, &deviceApplyPendingReply{} + case "ClearShadow": + return &deviceClearShadowRequest{}, &deviceClearShadowReply{} case "Configure": return &deviceConfigureRequest{}, &deviceConfigureReply{} case "MarkOffline": return &deviceMarkOfflineRequest{}, &deviceMarkOfflineReply{} case "Report": return &deviceReportRequest{}, &deviceReportReply{} + case "ReportAction": + return &deviceReportActionRequest{}, &deviceReportActionReply{} case "Shadow": return &deviceShadowRequest{}, &deviceShadowReply{} + case "ShadowExists": + return &deviceShadowExistsRequest{}, &deviceShadowExistsReply{} default: return nil, nil } @@ -110,6 +177,83 @@ func newDeviceProxy(rt gor.Invoker, id gor.GrainId) domain.Device { return &deviceProxy{id: id, rt: rt} } +type recoveryCoordinatorProxy struct { + id gor.GrainId + rt gor.Invoker +} + +type recoveryCoordinatorRecoverRequest struct { + A0 gor.TickStatus +} +type recoveryCoordinatorRecoverReply struct{} + +func (p *recoveryCoordinatorProxy) Recover(ctx context.Context, tick gor.TickStatus) error { + var reply recoveryCoordinatorRecoverReply + err := p.rt.Invoke(ctx, p.id, "Recover", &recoveryCoordinatorRecoverRequest{A0: tick}, &reply) + return err +} + +type recoveryCoordinatorStartRequest struct{} +type recoveryCoordinatorStartReply struct{} + +func (p *recoveryCoordinatorProxy) Start(ctx context.Context) error { + var reply recoveryCoordinatorStartReply + err := p.rt.Invoke(ctx, p.id, "Start", &recoveryCoordinatorStartRequest{}, &reply) + return err +} + +type recoveryCoordinatorStopRequest struct{} +type recoveryCoordinatorStopReply struct{} + +func (p *recoveryCoordinatorProxy) Stop(ctx context.Context) error { + var reply recoveryCoordinatorStopReply + err := p.rt.Invoke(ctx, p.id, "Stop", &recoveryCoordinatorStopRequest{}, &reply) + return err +} + +func dispatchRecoveryCoordinator(ctx context.Context, instance domain.RecoveryCoordinator, method string, args any, reply any) error { + switch method { + case "Recover": + typedArgs := args.(*recoveryCoordinatorRecoverRequest) + err := instance.Recover(ctx, typedArgs.A0) + return err + case "Start": + err := instance.Start(ctx) + return err + case "Stop": + err := instance.Stop(ctx) + return err + default: + return fmt.Errorf("unknown method %q", method) + } +} + +func newRecoveryCoordinatorCall(method string) (args any, reply any) { + switch method { + case "Recover": + return &recoveryCoordinatorRecoverRequest{}, &recoveryCoordinatorRecoverReply{} + case "Start": + return &recoveryCoordinatorStartRequest{}, &recoveryCoordinatorStartReply{} + case "Stop": + return &recoveryCoordinatorStopRequest{}, &recoveryCoordinatorStopReply{} + default: + return nil, nil + } +} + +func newRecoveryCoordinatorReminderCall(method string, status gor.TickStatus) (args any, reply any) { + switch method { + case "Recover": + return &recoveryCoordinatorRecoverRequest{A0: status}, &recoveryCoordinatorRecoverReply{} + default: + return nil, nil + } +} + +func newRecoveryCoordinatorProxy(rt gor.Invoker, id gor.GrainId) domain.RecoveryCoordinator { + return &recoveryCoordinatorProxy{id: id, rt: rt} +} + type workshopProxy struct { id gor.GrainId rt gor.Invoker @@ -200,6 +344,9 @@ func Install(rt *gor.Runtime) error { if err := gor.InstallType[domain.Device](rt, dispatchDevice, newDeviceProxy, newDeviceCall, newDeviceReminderCall); err != nil { return err } + if err := gor.InstallType[domain.RecoveryCoordinator](rt, dispatchRecoveryCoordinator, newRecoveryCoordinatorProxy, newRecoveryCoordinatorCall, newRecoveryCoordinatorReminderCall); err != nil { + return err + } if err := gor.InstallType[domain.Workshop](rt, dispatchWorkshop, newWorkshopProxy, newWorkshopCall, newWorkshopReminderCall); err != nil { return err } diff --git a/examples/shadow/domain/recovery.go b/examples/shadow/domain/recovery.go new file mode 100644 index 0000000..e5161d1 --- /dev/null +++ b/examples/shadow/domain/recovery.go @@ -0,0 +1,76 @@ +package domain + +import ( + "context" + "errors" + + "github.com/suraciii/gor" +) + +type recoveryCoordinator struct { + binder *gor.Binder + status gor.State[CoordinatorState] + reminder gor.Reminder[RecoveryCoordinator] + application ApplicationStore + observed chan<- RecoveryObservation +} + +// NewRecoveryCoordinator creates the fixed-key recovery Grain. +func NewRecoveryCoordinator(b *gor.Binder, application ApplicationStore) RecoveryCoordinator { + return newRecoveryCoordinator(b, application, nil) +} + +// NewRecoveryCoordinatorWithObservation creates the recovery Grain and sends +// each Recover context and tick to observed. The channel is for example tests. +func NewRecoveryCoordinatorWithObservation(b *gor.Binder, application ApplicationStore, observed chan<- RecoveryObservation) RecoveryCoordinator { + return newRecoveryCoordinator(b, application, observed) +} + +func newRecoveryCoordinator(b *gor.Binder, application ApplicationStore, observed chan<- RecoveryObservation) RecoveryCoordinator { + return &recoveryCoordinator{ + binder: b, + status: gor.NewState[CoordinatorState](b, "running"), + reminder: gor.NewReminder[RecoveryCoordinator](b), + application: application, + observed: observed, + } +} + +func (c *recoveryCoordinator) Start(ctx context.Context) error { + if c.status.Exists() && c.status.Get().Running { + return nil + } + if err := c.reminder.Set(ctx, RecoveryReminderName, gor.Every(RecoveryInterval), gor.Handle(RecoveryCoordinator.Recover)); err != nil { + return err + } + return c.status.Set(ctx, CoordinatorState{Running: true}) +} + +func (c *recoveryCoordinator) Stop(ctx context.Context) error { + // Clear State first. If the process stops before Cancel completes, Start + // sees a stopped coordinator and can safely restore the Reminder. + if err := c.status.Clear(ctx); err != nil { + return err + } + return c.reminder.Cancel(ctx, RecoveryReminderName) +} + +func (c *recoveryCoordinator) Recover(ctx context.Context, tick gor.TickStatus) error { + if c.observed != nil { + traceID, present := gor.RequestContextValue(ctx, "trace_id") + c.observed <- RecoveryObservation{Tick: tick, TraceID: traceID, TracePresent: present} + } + if c.application == nil { + return errors.New("application store is not configured") + } + actions, err := c.application.ListPending(ctx) + if err != nil { + return err + } + for _, action := range actions { + if err := gor.Ref[Device](c.binder, action.DeviceKey).ApplyPending(ctx, action.ActionID); err != nil { + return err + } + } + return nil +} diff --git a/examples/shadow/runtime.go b/examples/shadow/runtime.go index de2333f..7905c61 100644 --- a/examples/shadow/runtime.go +++ b/examples/shadow/runtime.go @@ -32,6 +32,35 @@ func RegisterWithLifecycle(rt *gor.Runtime, events chan<- domain.LifecycleEvent) ) } +// RegisterConformance installs the Single Silo recovery example with an +// application-owned store. The runtime uses no membership or transport. +func RegisterConformance(rt *gor.Runtime, application domain.ApplicationStore) error { + return registerConformance(rt, application, nil) +} + +// RegisterConformanceWithObservation installs the conformance example and +// sends Reminder observations to observed for deterministic tests. +func RegisterConformanceWithObservation(rt *gor.Runtime, application domain.ApplicationStore, observed chan<- domain.RecoveryObservation) error { + return registerConformance(rt, application, observed) +} + +func registerConformance(rt *gor.Runtime, application domain.ApplicationStore, observed chan<- domain.RecoveryObservation) error { + if err := register(rt, + func(b *gor.Binder) domain.Device { + return domain.NewDeviceWithApplication(b, application) + }, + domain.NewWorkshop, + ); err != nil { + return err + } + return gor.Register[domain.RecoveryCoordinator](rt, func(b *gor.Binder) domain.RecoveryCoordinator { + if observed == nil { + return domain.NewRecoveryCoordinator(b, application) + } + return domain.NewRecoveryCoordinatorWithObservation(b, application, observed) + }) +} + func register(rt *gor.Runtime, deviceFactory func(*gor.Binder) domain.Device, workshopFactory func(*gor.Binder) domain.Workshop) error { if err := gorgen.Install(rt); err != nil { return err diff --git a/store/sqlite.go b/store/sqlite.go index 7cfac64..f2db86b 100644 --- a/store/sqlite.go +++ b/store/sqlite.go @@ -271,7 +271,7 @@ func sqliteDSN(path string, durability Durability) string { if durability == DurabilityRelaxed { sync = "NORMAL" } - return fmt.Sprintf("file:%s?_pragma=journal_mode(WAL)&_pragma=synchronous(%s)&_pragma=busy_timeout(%d)", path, sync, sqliteBusyTimeout) + return fmt.Sprintf("%s?_pragma=journal_mode(WAL)&_pragma=synchronous(%s)&_pragma=busy_timeout(%d)", sqliteFileURI(path), sync, sqliteBusyTimeout) } func stateFilePath(path string) string { diff --git a/store/sqlite_test.go b/store/sqlite_test.go index 2ca4bf1..7ee0162 100644 --- a/store/sqlite_test.go +++ b/store/sqlite_test.go @@ -3,6 +3,7 @@ package store import ( "context" "errors" + "os" "path/filepath" "strings" "testing" @@ -180,6 +181,42 @@ func TestSQLiteStore_PersistsAcrossReopen(t *testing.T) { } } +func TestOpenSQLite_PathWithURICharactersUsesRequestedFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "store?#.db") + id := GrainId{GrainType: "account", GrainKey: "uri"} + + first, err := OpenSQLite(path) + if err != nil { + t.Fatalf("OpenSQLite first: %v", err) + } + if _, err := first.Write(context.Background(), id, []byte("uri"), 0); err != nil { + first.Close() + t.Fatalf("Write: %v", err) + } + if err := first.Close(); err != nil { + t.Fatalf("Close first: %v", err) + } + if _, err := os.Stat(path); err != nil { + t.Fatalf("requested coordination database %q: %v", path, err) + } + if _, err := os.Stat(stateFilePath(path)); err != nil { + t.Fatalf("requested State database %q: %v", stateFilePath(path), err) + } + + second, err := OpenSQLite(path) + if err != nil { + t.Fatalf("OpenSQLite second: %v", err) + } + defer second.Close() + record, err := second.Read(context.Background(), id) + if err != nil { + t.Fatalf("Read after reopen: %v", err) + } + if string(record.Data) != "uri" { + t.Fatalf("Record after reopen = %#v, want uri", record) + } +} + func TestOpenSQLite_MissingParentDirErrorNamesPath(t *testing.T) { path := filepath.Join(t.TempDir(), "no", "such", "dir", "gor.db")