feat: 接入 Instance 版本与扩展可用性观测
This commit is contained in:
@@ -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}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user