Author SHA1 Message Date
panxiao81 e2e795a889 Merge pull request 'feat: 接入 Instance 版本与扩展可用性观测' (#8) from feat/database-metadata-observation into main
Verify / test (push) Successful in 10m53s
Verify / lint (push) Successful in 12m26s
Verify / database-integration (push) Successful in 16m44s
2026-09-24 16:51:48 +00:00
panxiao81 9441b568da feat: 接入 Instance 版本与扩展可用性观测
Verify / test (pull_request) Successful in 6m1s
Verify / lint (pull_request) Successful in 7m19s
Verify / database-integration (pull_request) Successful in 5m11s
2026-09-24 15:14:43 +00:00
panxiao81 6acba0ca46 Merge pull request 'feat: 迁移 PostgreSQL registry 所有权存储' (#7) from feat/database-registry into main
Verify / test (push) Successful in 8m36s
Verify / lint (push) Successful in 10m32s
Verify / database-integration (push) Successful in 6m9s
2026-09-24 14:46:52 +00:00
panxiao81 77848c7dc8 Merge pull request 'feat: 接入管理 Secret 凭据与连接刷新' (#6) from feat/database-admin-credentials into main
Verify / test (push) Successful in 5m45s
Verify / database-integration (push) Successful in 11m41s
Verify / lint (push) Successful in 12m38s
2026-09-24 14:45:58 +00:00
panxiao81 e590542b0b feat: 迁移 PostgreSQL registry 所有权存储与恢复测试
Verify / test (pull_request) Successful in 9m18s
Verify / lint (pull_request) Successful in 10m19s
Verify / database-integration (pull_request) Successful in 11m32s
2026-09-21 15:59:14 +00:00
17 changed files with 1265 additions and 52 deletions
+46 -2
View File
@@ -46,8 +46,9 @@ Secret metadata 和无关字段变化不重建连接。观测后再次读取 Sec
不把旧连接的成功作为新凭据有效的证据;这不构成跨 Kubernetes/PostgreSQL 的原子事务。
Instance UID、endpoint 或凭据引用变化也会释放旧连接;Forget/Close 只释放本地资源。
当前只有 `ObserveVersion` 版本查询用例,不能产生完整 CapabilityObservation 或 Ready。
controller 接入、Secret watch、finalizer、registry 与真实权限检查仍待后续切片;并发 CR 更新
当前通过 `ObserveMetadata` 读取服务器版本和可用扩展,`ObserveVersion` 只是其版本读取便捷入口,
不能产生完整 CapabilityObservation 或 Ready。
controller 接入、Secret watch、finalizer、registry 观测装配与真实权限检查仍待后续切片;并发 CR 更新
必须由调用者通过 resourceVersion 校验。应用层沿用源实现的串行处理,本阶段未引入新的调度框架。
运行 `make test-database-integration` 验证真实 API server + 一次性 PostgreSQLfixture 不接受外部
@@ -58,6 +59,49 @@ namespace 边界、有效值轮换、metadata 无关变化、中途轮换、重
integration 标签代码。领域单测、真实 API 行为与真实 PostgreSQL 行为分别验收,不以本切片
替代整个 Instance controller 的集成验收。
## 服务器 metadata 与扩展观测
SQL adapter 通过一条只读语句读取 `pg_catalog.current_setting('server_version')`
`pg_catalog.pg_available_extensions`,避免从已安装列表推断可用列表,且不依赖可修改的
`search_path`。查询失败丢弃整份结果;成功返回空列表与尚未观察严格区分。
参考 PostgreSQL 的 [pg_available_extensions](https://www.postgresql.org/docs/18/view-pg-available-extensions.html)
与 [CREATE EXTENSION](https://www.postgresql.org/docs/18/sql-createextension.html) 合同:可用列表
表示服务器提供的扩展,不证明管理账号有安装权限,也不保证依赖和其他安装前提满足。
`InstanceService` 保留原有 CredentialReader → Connector → Database 边界,复用同一个
凭据读取、连接刷新和串行释放流程,不新增连接池封装或任意查询回调。每次重新查询 metadata,
并在 Secret 有效值回读一致后生成不可变的 `MetadataObservation`,绑定本次 target(含当前
generation),不绑定建池时的旧 target。结果不包含凭据,扩展集合不与 driver 的可变 slice 共享。
凭据中途变化、读取失败或查询失败时,返回零值观察并释放连接,不复用旧的扩展列表。
调用方可将 `Target()``Extensions()` 交给 Instance 的 `ObserveExtensions`;应用调用链
仍负责同轮次使用,不能持久化或跨轮缓存这份证据。metadata 读取不安装扩展、不初始化 registry、
不设置 Ready,也不授予 Tenant 写权限。管理权限矩阵、registry 完整回读及 controller 的
checkpoint/status/finalizer 链路仍是后续切片。
真实 API server + PostgreSQL 测试验证未安装扩展可被观察、名称保持大小写、search_path 遮蔽
不改变查询来源、低权限账号读取、权限撤回失败与恢复、Secret 中途变化丢弃扩展结果。
单元测试补充成功空列表、查询附带部分数据时丢弃、结果与可变 slice 隔离、每轮重新读取和
generation 变化时的目标绑定;原凭据/TLS/超时/并发测试沿同一 metadata 路径继续运行。
## Registry 所有权存储切片
`adapter/postgresql/registry` 直接迁入上述固定基线的 `internal/postgresql/registry`
保留原 schema、版本表和唯一约束,不接回旧 controller。Store 使用调用方提供的 pgx 连接或
连接池,不创建连接池、不读取凭据、不管理重连。迁移复用 tern v2.4.3 的事务和 advisory lock
所有权变更复用 PostgreSQL 唯一约束与行锁,没有新增迁移或锁框架。
Claim 支持相同归属的幂等重试;Retain 保留不可重新占用的墓碑;Delete 仅删除匹配归属的
managed 记录,且只应在调用方确认外部资源已删除后调用。Store 本身不删除 database、role
或凭据,也不授予 Tenant provisioning 权限。驱动错误仅供内部调用方分类,接入 controller 时
仍须按安全合同转换为脱敏的 Condition/Event,不能直接发布原始错误。
真实 PostgreSQL 测试覆盖重复/并发初始化、唯一约束冲突、并发 Claim、归属不匹配、Retain
与 Delete 重试、连接重建、取消恢复、未知 schema/超前版本拒绝,以及真实事务提交后注入
客户端失败的恢复分支。后者确定性模拟“提交成功但调用方不知道”,不声称覆盖真实网络分区。
registry 尚未装配到 InstanceService;初始化成功不能当作完整 registry 回读或 Instance Ready。
controller 重启、Secret watch、finalizer 与跨后端删除仍由后续 envtest/端到端切片验收。
## 设计入口
- [系统规格](specification.md):规范性行为与验收标准;
+12 -2
View File
@@ -3,7 +3,7 @@
> 本页迁入作为 Database 模块的测试分层与 fixture 合同。旧项目的 Make target、devcontainer
> 和脚手架版本尚未适配 Ayatori;实现时应复用 Ayatori 现有工具链,并保持这里定义的测试边界。
## Ayatori 已接入的凭据切片测试
## Ayatori 已接入的凭据、metadata 与 registry 切片测试
本节命令已在 Ayatori 接入;以下历史 Compose/Kind 操作仍属于迁入的目标合同。
@@ -28,7 +28,17 @@ runner 后端修复上线,不能从本地测试通过推断远端已经可用
fixture 启动失败会保留退出错误与 stderr,并遮蔽测试密码,
以区分缺少命令、daemon 不可达、权限和镜像拉取失败。
这些测试尚不包含 Instance CRD/controller、Secret watch、status/finalizer 事件链、registry
metadata 测试验证版本与可用扩展的只读查询,包括未安装扩展、大小写保持、search_path 遮蔽
低权限账号读取、catalog 访问被撤回后的失败与恢复,以及凭据中途变化时同时丢弃版本和扩展。
权限撤回只修改每个场景自建 PostgreSQL 容器的 ACL;不连接现有服务。
可用列表不等于安装权限,这些检查不替代后续的完整管理权限矩阵或 Instance Ready 验收。
registry 测试独立使用一次性 PostgreSQL,不启动 Kubernetes API server。覆盖重复和并发迁移、
所有权唯一约束、并发占用、Retain 墓碑与 Delete 幂等、连接重建、取消恢复、未知 schema 与
超前版本拒绝。通过在真实 COMMIT 成功后注入客户端错误,验证结果不确定时的重试;这不替代
真实网络故障测试,也不覆盖跨后端的删除步骤。
这些测试尚不包含 Instance CRD/controller、Secret watch、status/finalizer 事件链、registry 观测装配、
权限探测矩阵、ESO 或 Tenant 供应。版本查询成功不意味着 Instance Ready。
本项目同时依赖 Kubernetes API、PostgreSQL、OpenBao 和 ESO。日常开发不连接 homelab
+13 -2
View File
@@ -4,6 +4,7 @@ go 1.27.1
require (
github.com/jackc/pgx/v5 v5.11.0
github.com/jackc/tern/v2 v2.4.3
k8s.io/api v0.37.0
k8s.io/apimachinery v0.37.0
k8s.io/client-go v0.37.0
@@ -12,6 +13,10 @@ require (
require (
cel.dev/expr v0.25.1 // indirect
dario.cat/mergo v1.0.2 // indirect
github.com/Masterminds/goutils v1.1.1 // indirect
github.com/Masterminds/semver/v3 v3.5.0 // indirect
github.com/Masterminds/sprig/v3 v3.3.0 // indirect
github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/blang/semver/v4 v4.0.0 // indirect
@@ -44,11 +49,14 @@ require (
github.com/google/gnostic-models v0.7.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect
github.com/huandu/xstrings v1.5.0 // indirect
github.com/inconshreveable/mousetrap v1.1.0 // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
github.com/jackc/puddle/v2 v2.2.2 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/mitchellh/copystructure v1.2.0 // indirect
github.com/mitchellh/reflectwalk v1.0.2 // indirect
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
@@ -57,6 +65,8 @@ require (
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.70.0 // indirect
github.com/prometheus/procfs v0.21.1 // indirect
github.com/shopspring/decimal v1.4.0 // indirect
github.com/spf13/cast v1.10.0 // indirect
github.com/spf13/cobra v1.10.2 // indirect
github.com/spf13/pflag v1.0.10 // indirect
github.com/x448/float16 v0.8.4 // indirect
@@ -72,14 +82,15 @@ require (
go.uber.org/multierr v1.11.0 // indirect
go.uber.org/zap v1.27.1 // indirect
go.yaml.in/yaml/v2 v2.4.4 // indirect
go.yaml.in/yaml/v3 v3.0.4 // indirect
go.yaml.in/yaml/v3 v3.0.5 // indirect
golang.org/x/crypto v0.55.0 // indirect
golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f // indirect
golang.org/x/net v0.57.0 // indirect
golang.org/x/oauth2 v0.36.0 // indirect
golang.org/x/sync v0.22.0 // indirect
golang.org/x/sys v0.47.0 // indirect
golang.org/x/term v0.45.0 // indirect
golang.org/x/text v0.40.0 // indirect
golang.org/x/text v0.41.0 // indirect
golang.org/x/time v0.15.0 // indirect
gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
+34 -13
View File
@@ -1,7 +1,13 @@
cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4=
cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4=
github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8=
dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
github.com/Masterminds/goutils v1.1.1 h1:5nUrii3FMTL5diU80unEVvNevw1nH4+ZV4DSLVJLSYI=
github.com/Masterminds/goutils v1.1.1/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=
github.com/Masterminds/semver/v3 v3.5.0 h1:kQceYJfbupGfZOKZQg0kou0DgAKhzDg2NZPAwZ/2OOE=
github.com/Masterminds/semver/v3 v3.5.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
github.com/Masterminds/sprig/v3 v3.3.0 h1:mQh0Yrg1XPo6vjYXgtf5OtijNAKJRNcTdOOGZe3tPhs=
github.com/Masterminds/sprig/v3 v3.3.0/go.mod h1:Zy1iXRYNqNLUolqCpL4uhk6SHUMAOSCzdgBfDb35Lz0=
github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -25,6 +31,8 @@ github.com/evanphx/json-patch/v5 v5.9.11 h1:/8HVnzMq13/3x9TPvjG08wUGqBTmZBsCWzjT
github.com/evanphx/json-patch/v5 v5.9.11/go.mod h1:3j+LviiESTElxA4p3EMKAB9HXj3/XEtnUf6OZxqIQTM=
github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ=
@@ -89,6 +97,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs=
github.com/huandu/xstrings v1.5.0 h1:2ag3IFq9ZDANvthTwTiqSSZLjDc+BedvHPAp5tJy2TI=
github.com/huandu/xstrings v1.5.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
@@ -99,6 +109,8 @@ github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg=
github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
github.com/jackc/tern/v2 v2.4.3 h1:g293d3OZgW7OFEhsYXgEv0C21jea2boNr0VR4K5I7OY=
github.com/jackc/tern/v2 v2.4.3/go.mod h1:rMpMuRYcff5wWLptoTSO1qcDxJ4OodysvK17i2SVBys=
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
github.com/klauspost/compress v1.19.0 h1:sXLILfc9jV2QYWkzFOPWStmcUVH2RHEB1JCdY2oVvCQ=
@@ -109,6 +121,10 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw=
github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ=
github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -137,6 +153,10 @@ github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIj
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY=
github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo=
github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU=
github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4=
github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
@@ -147,8 +167,8 @@ github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4=
github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
@@ -179,12 +199,15 @@ go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc=
go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=
go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M=
golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis=
golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f h1:W3F4c+6OLc6H2lb//N1q4WpJkhzJCK5J6kUi1NTVXfM=
golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f/go.mod h1:J1xhfL/vlindoeF/aINzNzt2Bket5bjo9sdOYzOsU80=
golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ=
golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0=
golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk=
golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40=
golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
@@ -195,12 +218,12 @@ golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0=
golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w=
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8=
golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q=
golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA=
golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE=
golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk=
gomodules.xyz/jsonpatch/v2 v2.4.0 h1:Ci3iUJyx9UeRx7CeFN8ARgGbkESwJK+KB9lLcWxY/Zw=
gomodules.xyz/jsonpatch/v2 v2.4.0/go.mod h1:AH3dM2RI6uoBZxn3LVrfvJ3E0/9dG4cSrbuBJT4moAY=
gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
@@ -214,8 +237,6 @@ google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3
google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af h1:+5/Sw3GsDNlEmu7TfklWKPdQ0Ykja5VEmq2i817+jbI=
google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnfEbYzo=
gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M=
gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
@@ -47,14 +47,6 @@ func (d *database) Close() {
d.pool.Close()
}
func (d *database) Version(ctx context.Context) (string, error) {
var version string
if err := d.pool.QueryRow(ctx, "SHOW server_version").Scan(&version); err != nil {
return "", safeError(err, application.ErrObservation)
}
return version, nil
}
func (c Connector) Connect(ctx context.Context, endpoint instance.Endpoint, credentials application.Credentials) (application.Database, error) {
if err := endpoint.Validate(); err != nil {
return nil, err
@@ -131,13 +131,17 @@ func TestObservationDiscardsResultWhenCredentialsChange(t *testing.T) {
}
}
version, err := fixture.service.ObserveVersion(fixture.ctx, fixture.target)
observation, err := fixture.service.ObserveMetadata(fixture.ctx, fixture.target)
if !errors.Is(err, application.ErrCredentialsChanged) {
t.Fatal("in-flight rotation was not detected")
}
if version != "" {
if observation.Version() != "" {
t.Fatal("observation returned data obtained with stale credentials")
}
requested := instance.NewExtensionSet([]string{fixtureExtension})
if observation.Extensions().Check(requested).Decision != instance.ExtensionSupportUnobserved {
t.Fatal("observation returned extension support obtained with stale credentials")
}
if fixture.backendIDs(t) != "" {
t.Fatal("stale connection was retained after rotation")
}
@@ -42,6 +42,7 @@ import (
const (
fixtureHost = "fixture.invalid"
fixtureUser = "postgres"
fixtureExtension = "plpgsql"
dockerExec = "exec"
fixtureImage = "postgres@sha256:18cfe3ef5e6815560c98237d6216d1e5119702fb0f3894c8785dd58b8bbe5d73"
fixturePassword = "AYATORI-TEST-ONLY-initial-password"
@@ -0,0 +1,48 @@
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package postgresql
import (
"context"
"git.ddupan.top/panxiao81/ayatori/internal/database/application"
)
// 使用 pg_catalog 限定名称,避免管理账号的 search_path 改变查询来源。
// 一个语句读取版本和可用列表;ARRAY 子查询在无行时返回空数组,而非 NULL。
// 不查询 pg_extension:已安装集合不能代表服务器提供的全部扩展。
const inspectMetadataStatement = `
SELECT
pg_catalog.current_setting('server_version'),
ARRAY(
SELECT name::text
FROM pg_catalog.pg_available_extensions
ORDER BY name
)`
func (d *database) InspectMetadata(ctx context.Context) (application.DatabaseMetadata, error) {
var metadata application.DatabaseMetadata
err := d.pool.QueryRow(ctx, inspectMetadataStatement).Scan(
&metadata.Version,
&metadata.AvailableExtensions,
)
if err != nil {
// 不返回部分结果,也不把查询失败转换为“成功观察到空列表”。
return application.DatabaseMetadata{}, safeError(err, application.ErrObservation)
}
return metadata, nil
}
@@ -0,0 +1,115 @@
//go:build integration
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package postgresql_test
import (
"errors"
"testing"
corev1 "k8s.io/api/core/v1"
"git.ddupan.top/panxiao81/ayatori/internal/database/application"
"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
)
func TestMetadataObservesAvailableExtensionsWithoutInstalling(t *testing.T) {
f := newCredentialFixture(t)
f.createSecret(t, controllerNamespace)
if installed := f.queryPostgres(t, "SELECT count(*) FROM pg_catalog.pg_extension WHERE extname = 'hstore'"); installed != "0" {
t.Fatal("fixture unexpectedly has hstore installed")
}
// 提供同名遮蔽对象,验证 adapter 不依赖管理账号可修改的 search_path。
f.queryPostgres(t, "CREATE VIEW public.pg_available_extensions AS SELECT 'fake_extension'::name AS name")
f.queryPostgres(t, "ALTER ROLE postgres SET search_path = public, pg_catalog")
observed, err := f.service.ObserveMetadata(f.ctx, f.target)
if err != nil {
t.Fatal(err)
}
if !observed.Target().Matches(f.target) || observed.Version() == "" {
t.Fatal("metadata was not bound to the current target")
}
requested := instance.NewExtensionSet([]string{"hstore", fixtureExtension})
if observed.Extensions().Check(requested).Decision != instance.ExtensionsAccepted {
t.Fatal("available but uninstalled extension was omitted")
}
unsupported := instance.NewExtensionSet([]string{"fake_extension", "HSTORE"})
check := observed.Extensions().Check(unsupported)
if check.Decision != instance.ExtensionsUnsupported || len(check.Unsupported) != 2 {
t.Fatal("metadata accepted shadowed or case-normalized extension names")
}
if installed := f.queryPostgres(t, "SELECT count(*) FROM pg_catalog.pg_extension WHERE extname = 'hstore'"); installed != "0" {
t.Fatal("metadata observation installed an extension")
}
if schemas := f.queryPostgres(t, "SELECT count(*) FROM pg_catalog.pg_namespace WHERE nspname = 'postgresql_tenant_operator'"); schemas != "0" {
t.Fatal("metadata observation initialized the registry")
}
aggregate, err := instance.Reconstitute(f.target, instance.Snapshot{}, false)
if err != nil {
t.Fatal(err)
}
if err := aggregate.ObserveExtensions(observed.Target(), observed.Extensions()); err != nil {
t.Fatal(err)
}
if aggregate.CheckExtensions(requested).Decision != instance.ExtensionsAccepted {
t.Fatal("domain rejected observed extension availability")
}
if err := aggregate.RequireProvisioningReady(); err == nil {
t.Fatal("extension availability incorrectly authorized provisioning")
}
}
func TestMetadataPermissionFailureAndRecovery(t *testing.T) {
f := newCredentialFixture(t)
f.createSecret(t, controllerNamespace)
// 低权限账号也能读取可用列表;这不能证明具备 role/database/extension 管理权限。
f.queryPostgres(t, "CREATE ROLE metadata_reader LOGIN PASSWORD '"+fixturePassword+"'")
f.updateSecret(t, func(secret *corev1.Secret) {
secret.Data["login"] = []byte("metadata_reader")
})
if flags := f.queryPostgres(t, "SELECT rolsuper, rolcreaterole, rolcreatedb FROM pg_catalog.pg_roles WHERE rolname = 'metadata_reader'"); flags != "f|f|f" {
t.Fatal("metadata reader unexpectedly has management privileges")
}
requested := instance.NewExtensionSet([]string{fixtureExtension})
observed, err := f.service.ObserveMetadata(f.ctx, f.target)
if err != nil || observed.Extensions().Check(requested).Decision != instance.ExtensionsAccepted {
t.Fatalf("read-only account could not observe metadata: %v", err)
}
// 仅操作本测试独占容器的 catalog ACL;失败不能转换成“不支持任何扩展”。
f.queryPostgres(t, "REVOKE SELECT ON pg_catalog.pg_available_extensions FROM PUBLIC")
failed, err := f.service.ObserveMetadata(f.ctx, f.target)
if !errors.Is(err, application.ErrObservation) {
t.Fatalf("metadata permission failure was not reported safely: %v", err)
}
if failed.Version() != "" || failed.Target().Validate() == nil {
t.Fatal("permission failure returned partial metadata")
}
if failed.Extensions().Check(requested).Decision != instance.ExtensionSupportUnobserved {
t.Fatal("permission failure returned an observed empty set")
}
if f.backendIDs(t) != "" {
t.Fatal("failed metadata connection was retained")
}
f.queryPostgres(t, "GRANT SELECT ON pg_catalog.pg_available_extensions TO PUBLIC")
recovered, err := f.service.ObserveMetadata(f.ctx, f.target)
if err != nil || recovered.Extensions().Check(requested).Decision != instance.ExtensionsAccepted {
t.Fatalf("metadata observation did not recover: %v", err)
}
}
@@ -0,0 +1,25 @@
CREATE SCHEMA postgresql_tenant_operator;
CREATE TABLE postgresql_tenant_operator.tenant_ownership (
instance_uid text NOT NULL,
tenant_uid text NOT NULL,
tenant_namespace text NOT NULL,
tenant_name text NOT NULL,
database_name text NOT NULL,
role_name text NOT NULL,
credential_path text NOT NULL,
managed boolean NOT NULL DEFAULT true,
created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
updated_at timestamptz NOT NULL DEFAULT clock_timestamp(),
retained_at timestamptz,
PRIMARY KEY (instance_uid, tenant_uid),
UNIQUE (instance_uid, tenant_namespace, tenant_name),
UNIQUE (instance_uid, database_name),
UNIQUE (instance_uid, role_name),
UNIQUE (credential_path),
CHECK (managed OR retained_at IS NOT NULL)
);
---- create above / drop below ----
DROP SCHEMA postgresql_tenant_operator CASCADE;
@@ -0,0 +1,326 @@
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package registry persists controller ownership evidence in the PostgreSQL
// management database. It deliberately does not persist reconciliation phases;
// those belong to the Kubernetes resource status.
package registry
import (
"context"
"embed"
"errors"
"fmt"
"io/fs"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/jackc/tern/v2/migrate"
)
const migrationVersionTable = "public.postgresql_tenant_operator_schema_version"
//go:embed migrations/*.sql
var migrationFiles embed.FS
var (
// ErrNotFound indicates that the registry has no matching ownership record.
ErrNotFound = errors.New("registry ownership record not found")
// ErrConflict indicates that a requested name or path belongs to another Tenant UID.
ErrConflict = errors.New("registry ownership conflict")
)
// Beginner is implemented by pgx.Conn and pgxpool.Pool.
type Beginner interface {
Begin(context.Context) (pgx.Tx, error)
}
// Store manages ownership records in one PostgreSQLInstance management database.
type Store struct {
db Beginner
}
// NewStore creates a registry store backed by a PostgreSQL connection or pool.
func NewStore(db Beginner) *Store {
return &Store{db: db}
}
// Ownership identifies every external resource reserved for one Tenant UID.
type Ownership struct {
InstanceUID string
TenantUID string
TenantNamespace string
TenantName string
DatabaseName string
RoleName string
CredentialPath string
}
// Record is the persisted ownership state.
type Record struct {
Ownership
Managed bool
CreatedAt time.Time
UpdatedAt time.Time
RetainedAt *time.Time
}
// ClaimResult reports whether Claim inserted a new record or found the same claim.
type ClaimResult string
const (
ClaimCreated ClaimResult = "Created"
ClaimOwned ClaimResult = "Owned"
)
// Bootstrap applies all pending versioned registry migrations idempotently.
func (s *Store) Bootstrap(ctx context.Context) error {
if s == nil || s.db == nil {
return errors.New("bootstrap registry: nil database")
}
return s.withMigrationConnection(ctx, func(conn *pgx.Conn) error {
migrations, err := fs.Sub(migrationFiles, "migrations")
if err != nil {
return fmt.Errorf("open embedded registry migrations: %w", err)
}
migrator, err := migrate.NewMigrator(ctx, conn, migrationVersionTable)
if err != nil {
return fmt.Errorf("initialize registry migrator: %w", err)
}
if err := migrator.LoadMigrations(migrations); err != nil {
return fmt.Errorf("load registry migrations: %w", err)
}
if err := migrator.Migrate(ctx); err != nil {
return fmt.Errorf("apply registry migrations: %w", err)
}
return nil
})
}
// Claim reserves all names for an owner. Repeating an identical claim is idempotent.
func (s *Store) Claim(ctx context.Context, owner Ownership) (ClaimResult, error) {
if s == nil || s.db == nil {
return "", errors.New("claim registry ownership: nil database")
}
if err := owner.validate(); err != nil {
return "", err
}
tx, err := s.db.Begin(ctx)
if err != nil {
return "", fmt.Errorf("begin registry claim: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
_, err = scanRecord(tx.QueryRow(ctx, claimStatement,
owner.InstanceUID,
owner.TenantUID,
owner.TenantNamespace,
owner.TenantName,
owner.DatabaseName,
owner.RoleName,
owner.CredentialPath,
))
if err != nil && !errors.Is(err, ErrNotFound) {
return "", fmt.Errorf("insert registry claim: %w", err)
}
if errors.Is(err, ErrNotFound) {
record, err := getByTenantUID(ctx, tx, owner.InstanceUID, owner.TenantUID)
if err != nil {
if errors.Is(err, ErrNotFound) {
return "", fmt.Errorf("%w: database, role, tenant identity, or credential path is already reserved", ErrConflict)
}
return "", err
}
if !record.equal(owner) || !record.Managed {
return "", fmt.Errorf("%w: database, role, tenant identity, or credential path is already reserved", ErrConflict)
}
if err := tx.Commit(ctx); err != nil {
return "", fmt.Errorf("commit registry claim: %w", err)
}
return ClaimOwned, nil
}
if err := tx.Commit(ctx); err != nil {
return "", fmt.Errorf("commit registry claim: %w", err)
}
return ClaimCreated, nil
}
func (s *Store) withMigrationConnection(ctx context.Context, run func(*pgx.Conn) error) error {
switch db := s.db.(type) {
case *pgx.Conn:
return run(db)
case *pgxpool.Pool:
conn, err := db.Acquire(ctx)
if err != nil {
return fmt.Errorf("acquire registry migration connection: %w", err)
}
defer conn.Release()
return run(conn.Conn())
default:
return fmt.Errorf("bootstrap registry: database type %T cannot provide a migration connection", s.db)
}
}
// Get returns the ownership record for an Instance UID and Tenant UID.
func (s *Store) Get(ctx context.Context, instanceUID, tenantUID string) (Record, error) {
if s == nil || s.db == nil {
return Record{}, errors.New("get registry record: nil database")
}
if instanceUID == "" || tenantUID == "" {
return Record{}, errors.New("get registry record: instance UID and tenant UID are required")
}
tx, err := s.db.Begin(ctx)
if err != nil {
return Record{}, fmt.Errorf("begin registry read: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
record, err := getByTenantUID(ctx, tx, instanceUID, tenantUID)
if err != nil {
return Record{}, err
}
if err := tx.Commit(ctx); err != nil {
return Record{}, fmt.Errorf("commit registry read: %w", err)
}
return record, nil
}
// MarkRetained changes a matching managed ownership record into an unmanaged tombstone.
// Repeating the operation for the same tombstone is safe.
func (s *Store) MarkRetained(ctx context.Context, owner Ownership) error {
return s.changeOwnership(ctx, owner, "mark registry record retained", func(ctx context.Context, tx pgx.Tx, record Record) error {
if !record.Managed {
return nil
}
tag, err := tx.Exec(ctx, markRetainedStatement, owner.InstanceUID, owner.TenantUID)
if err != nil {
return err
}
if tag.RowsAffected() != 1 {
return ErrConflict
}
return nil
})
}
// Delete removes a matching managed record after its external resources have been deleted.
// An already absent record is treated as a successful retry; a retained record is never deleted.
func (s *Store) Delete(ctx context.Context, owner Ownership) error {
return s.changeOwnership(ctx, owner, "delete registry record", func(ctx context.Context, tx pgx.Tx, record Record) error {
if !record.Managed {
return ErrConflict
}
tag, err := tx.Exec(ctx, deleteStatement, owner.InstanceUID, owner.TenantUID)
if err != nil {
return err
}
if tag.RowsAffected() != 1 {
return ErrConflict
}
return nil
})
}
func (s *Store) changeOwnership(
ctx context.Context,
owner Ownership,
operation string,
change func(context.Context, pgx.Tx, Record) error,
) error {
if s == nil || s.db == nil {
return fmt.Errorf("%s: nil database", operation)
}
if err := owner.validate(); err != nil {
return fmt.Errorf("%s: %w", operation, err)
}
tx, err := s.db.Begin(ctx)
if err != nil {
return fmt.Errorf("begin %s: %w", operation, err)
}
defer func() { _ = tx.Rollback(ctx) }()
record, err := getByTenantUIDForUpdate(ctx, tx, owner.InstanceUID, owner.TenantUID)
if errors.Is(err, ErrNotFound) && operation == "delete registry record" {
return nil
}
if err != nil {
return fmt.Errorf("%s: %w", operation, err)
}
if !record.equal(owner) {
return fmt.Errorf("%s: %w", operation, ErrConflict)
}
if err := change(ctx, tx, record); err != nil {
return fmt.Errorf("%s: %w", operation, err)
}
if err := tx.Commit(ctx); err != nil {
return fmt.Errorf("commit %s: %w", operation, err)
}
return nil
}
func getByTenantUIDForUpdate(ctx context.Context, tx pgx.Tx, instanceUID, tenantUID string) (Record, error) {
return scanRecord(tx.QueryRow(ctx, getByTenantUIDForUpdateStatement, instanceUID, tenantUID))
}
func getByTenantUID(ctx context.Context, tx pgx.Tx, instanceUID, tenantUID string) (Record, error) {
return scanRecord(tx.QueryRow(ctx, getByTenantUIDStatement, instanceUID, tenantUID))
}
type rowScanner interface {
Scan(...any) error
}
func scanRecord(row rowScanner) (Record, error) {
var record Record
err := row.Scan(
&record.InstanceUID,
&record.TenantUID,
&record.TenantNamespace,
&record.TenantName,
&record.DatabaseName,
&record.RoleName,
&record.CredentialPath,
&record.Managed,
&record.CreatedAt,
&record.UpdatedAt,
&record.RetainedAt,
)
if errors.Is(err, pgx.ErrNoRows) {
return Record{}, ErrNotFound
}
if err != nil {
return Record{}, fmt.Errorf("read registry record: %w", err)
}
return record, nil
}
func (o Ownership) validate() error {
if o.InstanceUID == "" || o.TenantUID == "" || o.TenantNamespace == "" || o.TenantName == "" ||
o.DatabaseName == "" || o.RoleName == "" || o.CredentialPath == "" {
return errors.New("claim registry ownership: all ownership fields are required")
}
return nil
}
func (o Ownership) equal(other Ownership) bool {
return o == other
}
@@ -0,0 +1,68 @@
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package registry
const claimStatement = `
INSERT INTO postgresql_tenant_operator.tenant_ownership (
instance_uid,
tenant_uid,
tenant_namespace,
tenant_name,
database_name,
role_name,
credential_path
) VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT DO NOTHING
RETURNING
instance_uid,
tenant_uid,
tenant_namespace,
tenant_name,
database_name,
role_name,
credential_path,
managed,
created_at,
updated_at,
retained_at`
const getByTenantUIDStatement = `
SELECT
instance_uid,
tenant_uid,
tenant_namespace,
tenant_name,
database_name,
role_name,
credential_path,
managed,
created_at,
updated_at,
retained_at
FROM postgresql_tenant_operator.tenant_ownership
WHERE instance_uid = $1 AND tenant_uid = $2`
const getByTenantUIDForUpdateStatement = getByTenantUIDStatement + ` FOR UPDATE`
const markRetainedStatement = `
UPDATE postgresql_tenant_operator.tenant_ownership
SET managed = false, retained_at = clock_timestamp(), updated_at = clock_timestamp()
WHERE instance_uid = $1 AND tenant_uid = $2 AND managed = true`
const deleteStatement = `
DELETE FROM postgresql_tenant_operator.tenant_ownership
WHERE instance_uid = $1 AND tenant_uid = $2 AND managed = true`
@@ -0,0 +1,427 @@
//go:build integration
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package postgresql_test
import (
"context"
"errors"
"net"
"net/url"
"strconv"
"sync"
"testing"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/postgresql/registry"
)
// registryFixture 只连接本测试创建的容器,不接受外部数据库地址。
type registryFixture struct {
ctx context.Context
pool *pgxpool.Pool
store *registry.Store
}
func newRegistryFixture(t *testing.T) *registryFixture {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
t.Cleanup(cancel)
_, port := postgresFixture(t, ctx)
endpoint := url.URL{
Scheme: "postgres",
User: url.UserPassword(fixtureUser, fixturePassword),
Host: net.JoinHostPort("127.0.0.1", strconv.Itoa(port)),
Path: "/postgres",
RawQuery: "sslmode=disable",
}
pool, err := pgxpool.New(ctx, endpoint.String())
if err != nil {
t.Fatal("cannot configure registry fixture connection")
}
t.Cleanup(pool.Close)
if err := pool.Ping(ctx); err != nil {
t.Fatal("cannot connect to registry fixture")
}
return &registryFixture{ctx: ctx, pool: pool, store: registry.NewStore(pool)}
}
func registryOwner(suffix string) registry.Ownership {
return registry.Ownership{
InstanceUID: "instance-uid",
TenantUID: "tenant-uid-" + suffix,
TenantNamespace: "applications",
TenantName: "tenant-" + suffix,
DatabaseName: "database_" + suffix,
RoleName: "role_" + suffix,
CredentialPath: "credentials/" + suffix,
}
}
func (f *registryFixture) bootstrap(t *testing.T) {
t.Helper()
if err := f.store.Bootstrap(f.ctx); err != nil {
t.Fatalf("bootstrap registry: %v", err)
}
}
func (f *registryFixture) claim(t *testing.T, owner registry.Ownership, expected registry.ClaimResult) {
t.Helper()
result, err := f.store.Claim(f.ctx, owner)
if err != nil || result != expected {
t.Fatalf("claim: result=%q, error=%v; want %q", result, err, expected)
}
}
func TestRegistryOwnershipLifecycle(t *testing.T) {
f := newRegistryFixture(t)
f.bootstrap(t)
f.bootstrap(t)
owner := registryOwner("lifecycle")
f.claim(t, owner, registry.ClaimCreated)
f.claim(t, owner, registry.ClaimOwned)
record, err := f.store.Get(f.ctx, owner.InstanceUID, owner.TenantUID)
if err != nil {
t.Fatal(err)
}
if record.Ownership != owner || !record.Managed || record.CreatedAt.IsZero() || record.RetainedAt != nil {
t.Fatalf("unexpected ownership record: %+v", record)
}
// 错误的资源归属不能删除或 Retain 原记录。
wrongOwner := owner
wrongOwner.RoleName = "another_role"
if err := f.store.Delete(f.ctx, wrongOwner); !errors.Is(err, registry.ErrConflict) {
t.Fatalf("delete mismatched owner: %v", err)
}
if err := f.store.MarkRetained(f.ctx, wrongOwner); !errors.Is(err, registry.ErrConflict) {
t.Fatalf("retain mismatched owner: %v", err)
}
for range 2 {
if err := f.store.Delete(f.ctx, owner); err != nil {
t.Fatalf("delete retry: %v", err)
}
}
if _, err := f.store.Get(f.ctx, owner.InstanceUID, owner.TenantUID); !errors.Is(err, registry.ErrNotFound) {
t.Fatalf("read deleted record: %v", err)
}
}
func TestRegistryRetainedRecordCannotBeReclaimed(t *testing.T) {
f := newRegistryFixture(t)
f.bootstrap(t)
owner := registryOwner("retained")
f.claim(t, owner, registry.ClaimCreated)
if err := f.store.MarkRetained(f.ctx, owner); err != nil {
t.Fatal(err)
}
retained, err := f.store.Get(f.ctx, owner.InstanceUID, owner.TenantUID)
if err != nil {
t.Fatal(err)
}
if retained.Managed || retained.RetainedAt == nil {
t.Fatalf("missing retained tombstone: %+v", retained)
}
if err := f.store.MarkRetained(f.ctx, owner); err != nil {
t.Fatalf("retain retry: %v", err)
}
retried, err := f.store.Get(f.ctx, owner.InstanceUID, owner.TenantUID)
if err != nil || !retried.UpdatedAt.Equal(retained.UpdatedAt) {
t.Fatalf("retain retry changed tombstone: %+v, %v", retried, err)
}
if err := f.store.Delete(f.ctx, owner); !errors.Is(err, registry.ErrConflict) {
t.Fatalf("delete retained record: %v", err)
}
if _, err := f.store.Claim(f.ctx, owner); !errors.Is(err, registry.ErrConflict) {
t.Fatalf("reclaim retained record: %v", err)
}
owner.TenantUID = "replacement-uid"
if _, err := f.store.Claim(f.ctx, owner); !errors.Is(err, registry.ErrConflict) {
t.Fatalf("replacement tenant reclaimed tombstone: %v", err)
}
}
func TestRegistryUniqueReservations(t *testing.T) {
f := newRegistryFixture(t)
f.bootstrap(t)
owner := registryOwner("original")
f.claim(t, owner, registry.ClaimCreated)
tests := []struct {
name string
change func(*registry.Ownership)
}{
{name: "tenant UID", change: func(other *registry.Ownership) { other.TenantUID = owner.TenantUID }},
{name: "tenant name", change: func(other *registry.Ownership) { other.TenantName = owner.TenantName }},
{name: "database", change: func(other *registry.Ownership) { other.DatabaseName = owner.DatabaseName }},
{name: "role", change: func(other *registry.Ownership) { other.RoleName = owner.RoleName }},
{name: "credential path", change: func(other *registry.Ownership) { other.CredentialPath = owner.CredentialPath }},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
other := registryOwner("other")
tt.change(&other)
if _, err := f.store.Claim(f.ctx, other); !errors.Is(err, registry.ErrConflict) {
t.Fatalf("conflicting reservation: %v", err)
}
})
}
}
func TestRegistryConcurrentBootstrapAndClaims(t *testing.T) {
f := newRegistryFixture(t)
const workers = 8
start := make(chan struct{})
results := make(chan error, workers)
var group sync.WaitGroup
for range workers {
group.Go(func() {
<-start
results <- f.store.Bootstrap(f.ctx)
})
}
close(start)
group.Wait()
close(results)
for err := range results {
if err != nil {
t.Fatalf("concurrent bootstrap: %v", err)
}
}
owner := registryOwner("concurrent")
type claimOutcome struct {
result registry.ClaimResult
err error
}
claims := make(chan claimOutcome, workers)
start = make(chan struct{})
for range workers {
group.Go(func() {
<-start
result, err := f.store.Claim(f.ctx, owner)
claims <- claimOutcome{result: result, err: err}
})
}
close(start)
group.Wait()
close(claims)
created := 0
for outcome := range claims {
if outcome.err != nil {
t.Fatal(outcome.err)
}
switch outcome.result {
case registry.ClaimCreated:
created++
case registry.ClaimOwned:
default:
t.Fatalf("unexpected claim result: %q", outcome.result)
}
}
if created != 1 {
t.Fatalf("created %d records for the same owner; want 1", created)
}
}
func TestRegistryRestartAndCanceledRequestRecovery(t *testing.T) {
f := newRegistryFixture(t)
f.bootstrap(t)
owner := registryOwner("restart")
f.claim(t, owner, registry.ClaimCreated)
config := f.pool.Config()
f.pool.Close()
pool, err := pgxpool.NewWithConfig(f.ctx, config)
if err != nil {
t.Fatal("cannot reopen fixture connection")
}
t.Cleanup(pool.Close)
f.store = registry.NewStore(pool)
f.bootstrap(t)
// 模拟客户端丢失上次成功结果后,以新连接重试;归属证据必须来自数据库。
f.claim(t, owner, registry.ClaimOwned)
canceled, cancel := context.WithCancel(f.ctx)
cancel()
if _, err := f.store.Claim(canceled, registryOwner("canceled")); !errors.Is(err, context.Canceled) {
t.Fatalf("canceled claim: %v", err)
}
f.claim(t, registryOwner("canceled"), registry.ClaimCreated)
}
func TestRegistryBootstrapRejectsUnknownSchema(t *testing.T) {
f := newRegistryFixture(t)
// 人工创建的同名 schema 不能被初始化过程接管或覆盖。
if _, err := f.pool.Exec(f.ctx, "CREATE SCHEMA postgresql_tenant_operator"); err != nil {
t.Fatal(err)
}
if err := f.store.Bootstrap(f.ctx); err == nil {
t.Fatal("bootstrap accepted an unmanaged schema")
}
var version int
if err := f.pool.QueryRow(f.ctx, "SELECT version FROM public.postgresql_tenant_operator_schema_version").Scan(&version); err != nil {
t.Fatal(err)
}
if version != 0 {
t.Fatalf("failed migration advanced version to %d", version)
}
// 仅在本测试拥有的临时数据库中清除空冲突 schema,然后重试失败的迁移。
if _, err := f.pool.Exec(f.ctx, "DROP SCHEMA postgresql_tenant_operator"); err != nil {
t.Fatal(err)
}
f.bootstrap(t)
if _, err := f.pool.Exec(f.ctx, "UPDATE public.postgresql_tenant_operator_schema_version SET version = 100"); err != nil {
t.Fatal(err)
}
if err := f.store.Bootstrap(f.ctx); err == nil {
t.Fatal("bootstrap accepted a future migration version")
}
if err := f.pool.QueryRow(f.ctx, "SELECT version FROM public.postgresql_tenant_operator_schema_version").Scan(&version); err != nil {
t.Fatal(err)
}
if version != 100 {
t.Fatalf("bootstrap rewrote future version to %d", version)
}
}
// 数据库执行真实 COMMIT 后才注入客户端错误,模拟客户端无法确认提交结果。
// 这不是网络故障测试,但能确定性覆盖已提交、调用者却收到失败的恢复分支。
type lostCommitReply struct {
registry.Beginner
err error
}
func (db lostCommitReply) Begin(ctx context.Context) (pgx.Tx, error) {
tx, err := db.Beginner.Begin(ctx)
if err != nil {
return nil, err
}
return uncertainCommit{Tx: tx, err: db.err}, nil
}
type uncertainCommit struct {
pgx.Tx
err error
}
func (tx uncertainCommit) Commit(ctx context.Context) error {
if err := tx.Tx.Commit(ctx); err != nil {
return err
}
return tx.err
}
func TestRegistryRetriesAfterLostCommitReply(t *testing.T) {
f := newRegistryFixture(t)
f.bootstrap(t)
replyLost := errors.New("test-only lost commit reply")
uncertain := registry.NewStore(lostCommitReply{Beginner: f.pool, err: replyLost})
owner := registryOwner("uncertain")
if _, err := uncertain.Claim(f.ctx, owner); !errors.Is(err, replyLost) {
t.Fatalf("claim did not report lost reply: %v", err)
}
f.claim(t, owner, registry.ClaimOwned)
if err := uncertain.MarkRetained(f.ctx, owner); !errors.Is(err, replyLost) {
t.Fatalf("retain did not report lost reply: %v", err)
}
if err := f.store.MarkRetained(f.ctx, owner); err != nil {
t.Fatalf("retry uncertain retain: %v", err)
}
if err := f.store.Delete(f.ctx, owner); !errors.Is(err, registry.ErrConflict) {
t.Fatalf("uncertain retain lost tombstone protection: %v", err)
}
deletable := registryOwner("uncertain_delete")
f.claim(t, deletable, registry.ClaimCreated)
if err := uncertain.Delete(f.ctx, deletable); !errors.Is(err, replyLost) {
t.Fatalf("delete did not report lost reply: %v", err)
}
if err := f.store.Delete(f.ctx, deletable); err != nil {
t.Fatalf("retry uncertain delete: %v", err)
}
}
func TestRegistryConcurrentConflictingClaims(t *testing.T) {
f := newRegistryFixture(t)
f.bootstrap(t)
first := registryOwner("first")
second := registryOwner("second")
second.DatabaseName = first.DatabaseName
start := make(chan struct{})
results := make(chan error, 2)
var group sync.WaitGroup
for _, owner := range []registry.Ownership{first, second} {
group.Go(func() {
<-start
_, err := f.store.Claim(f.ctx, owner)
results <- err
})
}
close(start)
group.Wait()
close(results)
created, conflicts := 0, 0
for err := range results {
switch {
case err == nil:
created++
case errors.Is(err, registry.ErrConflict):
conflicts++
default:
t.Fatalf("unexpected concurrent claim error: %v", err)
}
}
if created != 1 || conflicts != 1 {
t.Fatalf("concurrent reservation: created=%d conflicts=%d; want one of each", created, conflicts)
}
}
func TestRegistryConcurrentRetainAndDelete(t *testing.T) {
f := newRegistryFixture(t)
f.bootstrap(t)
owner := registryOwner("delete_race")
f.claim(t, owner, registry.ClaimCreated)
start := make(chan struct{})
retainResult := make(chan error, 1)
deleteResult := make(chan error, 1)
go func() {
<-start
retainResult <- f.store.MarkRetained(f.ctx, owner)
}()
go func() {
<-start
deleteResult <- f.store.Delete(f.ctx, owner)
}()
close(start)
retainErr, deleteErr := <-retainResult, <-deleteResult
record, readErr := f.store.Get(f.ctx, owner.InstanceUID, owner.TenantUID)
switch {
case retainErr == nil:
// Retain 先取得行锁时,Delete 必须拒绝删除墓碑。
if !errors.Is(deleteErr, registry.ErrConflict) || readErr != nil || record.Managed || record.RetainedAt == nil {
t.Fatalf("retain won but tombstone was not protected: delete=%v read=%v record=%+v", deleteErr, readErr, record)
}
case errors.Is(retainErr, registry.ErrNotFound):
// Delete 先提交时,Retain 必须报告记录已不存在,不能重建墓碑。
if deleteErr != nil || !errors.Is(readErr, registry.ErrNotFound) {
t.Fatalf("delete won but record remains: delete=%v read=%v", deleteErr, readErr)
}
default:
t.Fatalf("unexpected retain/delete race: retain=%v delete=%v", retainErr, deleteErr)
}
}
@@ -107,7 +107,7 @@ func TestPostgreSQLTLSHostIdentity(t *testing.T) {
if err != nil {
t.Fatal("trusted DNS SAN connection failed", err)
}
if version, err := db.Version(ctx); err != nil || version == "" {
if metadata, err := db.InspectMetadata(ctx); err != nil || metadata.Version == "" {
db.Close()
t.Fatal("TLS metadata read failed", err)
}
@@ -33,9 +33,9 @@ var (
)
// Database 与 Connector 复用原项目 internal/instance/service.go 的能力边界。
// 版本查询只是本切片的连通性观察,不能产生领域 Ready。
// Metadata 只查询版本与可用扩展,不能产生领域 Ready。
type Database interface {
Version(context.Context) (string, error)
InspectMetadata(context.Context) (DatabaseMetadata, error)
Close()
}
@@ -74,19 +74,25 @@ func NewInstanceService(source CredentialReader, connector Connector) (*Instance
func (s *InstanceService) String() string { return "[redacted instance service]" }
func (s *InstanceService) GoString() string { return s.String() }
// ObserveVersion 返回当前目标和凭据下的版本;任何失败均返回空结果
// 调用者仍需使用 CR resourceVersion 保存前提防止 spec 并发修改;本方法不建立跨系统事务。
// ObserveVersion 是完整 metadata 读取的便捷入口,不再维护另一条连接或查询路径
func (s *InstanceService) ObserveVersion(ctx context.Context, target instance.ObservationTarget) (string, error) {
observation, err := s.ObserveMetadata(ctx, target)
return observation.Version(), err
}
// ObserveMetadata 返回当前目标和凭据下的版本与扩展;任何失败均丢弃全部结果。
// 调用者仍需使用 CR resourceVersion 保存前提防止 spec 并发修改;本方法不建立跨系统事务。
func (s *InstanceService) ObserveMetadata(ctx context.Context, target instance.ObservationTarget) (MetadataObservation, error) {
if err := target.Validate(); err != nil {
return "", err
return MetadataObservation{}, err
}
s.mu.Lock()
defer s.mu.Unlock()
if s.closed {
return "", ErrClosed
return MetadataObservation{}, ErrClosed
}
if err := ctx.Err(); err != nil {
return "", err
return MetadataObservation{}, err
}
// 先读取有效凭据。读取失败时不得继续使用缓存中的旧连接。
@@ -94,11 +100,11 @@ func (s *InstanceService) ObserveVersion(ctx context.Context, target instance.Ob
credentials, err := s.source.Read(ctx, target.Definition().AdminCredential())
if err != nil {
s.release(name)
return "", credentialError(err)
return MetadataObservation{}, credentialError(err)
}
if credentials.username == "" || credentials.password == "" {
s.release(name)
return "", ErrCredentialsInvalid
return MetadataObservation{}, ErrCredentialsInvalid
}
// 连接身份与有效值均未变化时复用 pgxpool;generation 本身不要求换池。
@@ -111,7 +117,7 @@ func (s *InstanceService) ObserveVersion(ctx context.Context, target instance.Ob
if current == nil {
database, err := s.connector.Connect(ctx, target.Definition().Endpoint(), credentials)
if err != nil {
return "", err
return MetadataObservation{}, err
}
current = &entry{
target: target,
@@ -121,23 +127,31 @@ func (s *InstanceService) ObserveVersion(ctx context.Context, target instance.Ob
s.entries[name] = current
}
version, err := current.database.Version(ctx)
metadata, err := current.database.InspectMetadata(ctx)
if err != nil {
s.release(name)
return "", err
return MetadataObservation{}, err
}
if metadata.Version == "" {
s.release(name)
return MetadataObservation{}, ErrObservation
}
// 回读后再检查凭据,避免把轮换前取得的结果交给新凭据的调用链。
latest, err := s.source.Read(ctx, target.Definition().AdminCredential())
if err != nil {
s.release(name)
return "", credentialError(err)
return MetadataObservation{}, credentialError(err)
}
if latest != credentials {
s.release(name)
return "", ErrCredentialsChanged
return MetadataObservation{}, ErrCredentialsChanged
}
return version, nil
return MetadataObservation{
target: target,
version: metadata.Version,
extensions: instance.ObserveExtensionSupport(metadata.AvailableExtensions),
}, nil
}
func credentialError(err error) error {
@@ -24,6 +24,8 @@ import (
"git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
)
const serviceTestPassword = "test-only"
// 延续源项目 Service 测试,用于穷举身份与装配失败;真实行为由 adapter 集成测试验证。
type sourceStub struct {
credentials Credentials
@@ -35,11 +37,14 @@ func (s *sourceStub) Read(context.Context, instance.CredentialReference) (Creden
}
type databaseStub struct {
closes int
err error
closes int
err error
metadata DatabaseMetadata
}
func (d *databaseStub) Version(context.Context) (string, error) { return "17", d.err }
func (d *databaseStub) InspectMetadata(context.Context) (DatabaseMetadata, error) {
return d.metadata, d.err
}
func (d *databaseStub) Close() {
d.closes++
}
@@ -53,7 +58,12 @@ func (c *connectorStub) Connect(context.Context, instance.Endpoint, Credentials)
if c.err != nil {
return nil, c.err
}
db := &databaseStub{}
db := &databaseStub{
metadata: DatabaseMetadata{
Version: "17",
AvailableExtensions: []string{"plpgsql"},
},
}
c.databases = append(c.databases, db)
return db, nil
}
@@ -98,7 +108,7 @@ func serviceTarget(t *testing.T, uid, host, secret string, generation int64) ins
}
func TestInstanceConnectionIdentity(t *testing.T) {
source := &sourceStub{credentials: Credentials{username: testUsername, password: "test-only"}}
source := &sourceStub{credentials: Credentials{username: testUsername, password: serviceTestPassword}}
connector := &connectorStub{}
service, err := NewInstanceService(source, connector)
if err != nil {
@@ -118,9 +128,13 @@ func TestInstanceConnectionIdentity(t *testing.T) {
{"same name with new UID", serviceTarget(t, "uid-2", "second", "replacement", 1), 4},
}
for _, testCase := range cases {
if _, err := service.ObserveVersion(ctx, testCase.target); err != nil {
observation, err := service.ObserveMetadata(ctx, testCase.target)
if err != nil {
t.Fatal(err)
}
if !observation.Target().Matches(testCase.target) {
t.Fatalf("%s: observation was bound to a previous target", testCase.name)
}
if len(connector.databases) != testCase.wantConnections {
t.Fatalf("%s: got %d connections, want %d", testCase.name, len(connector.databases), testCase.wantConnections)
}
@@ -137,11 +151,64 @@ func TestInstanceConnectionIdentity(t *testing.T) {
}
}
func TestMetadataObservationFreshnessAndFailure(t *testing.T) {
source := &sourceStub{credentials: Credentials{username: testUsername, password: serviceTestPassword}}
connector := &connectorStub{}
service, err := NewInstanceService(source, connector)
if err != nil {
t.Fatal(err)
}
defer service.Close()
ctx := context.Background()
target := serviceTarget(t, "metadata-uid", "first", "admin", 1)
observed, err := service.ObserveMetadata(ctx, target)
if err != nil || observed.Version() != "17" || !observed.Target().Matches(target) {
t.Fatalf("metadata observation: %v", err)
}
requested := instance.NewExtensionSet([]string{"plpgsql"})
if observed.Extensions().Check(requested).Decision != instance.ExtensionsAccepted {
t.Fatal("extension list was not observed")
}
// 连接可以复用,但每轮必须重新查询;旧观察还必须与 adapter 的可变 slice 脱离。
database := connector.databases[0]
database.metadata.AvailableExtensions[0] = "replacement"
if observed.Extensions().Check(requested).Decision != instance.ExtensionsAccepted {
t.Fatal("adapter mutation changed a completed observation")
}
refreshed, err := service.ObserveMetadata(ctx, target)
if err != nil || refreshed.Extensions().Check(requested).Decision != instance.ExtensionsUnsupported {
t.Fatalf("extension list was cached across observations: %v", err)
}
database.metadata.AvailableExtensions = nil
empty, err := service.ObserveMetadata(ctx, target)
if err != nil || empty.Extensions().Check(requested).Decision != instance.ExtensionsUnsupported {
t.Fatalf("successful empty list was treated as unobserved: %v", err)
}
// 即使 adapter 附带部分数据,错误仍使整个观察失效。
database.err = ErrObservation
failed, err := service.ObserveMetadata(ctx, target)
if !errors.Is(err, ErrObservation) || failed.Version() != "" || failed.Target().Validate() == nil {
t.Fatal("failed query returned a bound observation")
}
if failed.Extensions().Check(requested).Decision != instance.ExtensionSupportUnobserved {
t.Fatal("failed query was interpreted as an empty extension list")
}
if _, err := service.ObserveMetadata(ctx, target); err != nil {
t.Fatalf("retry after query failure: %v", err)
}
connector.databases[1].metadata.Version = ""
if _, err := service.ObserveMetadata(ctx, target); !errors.Is(err, ErrObservation) {
t.Fatal("missing server version was accepted as complete metadata")
}
}
func TestInstanceAssemblyFailureRecovery(t *testing.T) {
ctx := context.Background()
target := serviceTarget(t, "uid-1", "first", "admin", 1)
source := &sourceStub{
credentials: Credentials{username: testUsername, password: "test-only"},
credentials: Credentials{username: testUsername, password: serviceTestPassword},
err: errors.New("unsafe source error"),
}
connector := &connectorStub{err: ErrConnection}
+40
View File
@@ -0,0 +1,40 @@
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package application
import "git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance"
// DatabaseMetadata 是一次只读查询的事实,不包含管理权限或 registry 就绪结论。
// AvailableExtensions 是服务器提供的可用列表,不是已安装列表或安装授权。
type DatabaseMetadata struct {
Version string
AvailableExtensions []string
}
// MetadataObservation 只在查询成功且有效凭据再次核对一致后产生。
// target 绑定本次调用,而非连接最初创建时的 generation;零值表示没有观察。
type MetadataObservation struct {
target instance.ObservationTarget
version string
extensions instance.ExtensionSupport
}
func (o MetadataObservation) Target() instance.ObservationTarget { return o.target }
func (o MetadataObservation) Version() string { return o.version }
func (o MetadataObservation) Extensions() instance.ExtensionSupport {
return o.extensions
}