姚章浩
5eef13b615
方案A：所有对象写入经 loopback 网关，持久计划/grant/intent/逐版本 receipt 与 attempt→grant delivery 落在治理 PG；scope 由 provider+endpoint+bucket 决定， 同一物理桶只允许一个写域。版本 append 不等于 CAS，未锚定计划在首笔业务写前拒绝。 本提交只含网关与其适配层接线、协议文档；生产独占、凭据切换与桶设置未改动。
70 lines
2.2 KiB
Go
70 lines
2.2 KiB
Go
package objectgateway_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
g "knowledge-governance/internal/objectgateway"
|
|
)
|
|
|
|
type stoppedBeforeSend struct {
|
|
g.Backend
|
|
calls int
|
|
}
|
|
|
|
func (b *stoppedBeforeSend) Put(context.Context, string, []byte, string) (g.Version, error) {
|
|
b.calls++
|
|
return g.Version{}, errors.New("injected uncertain send with no visible version")
|
|
}
|
|
func TestGatewayAbsenceNeverSettlesUnknown(t *testing.T) {
|
|
ctx, pool, real := fixture(t)
|
|
backend := &stoppedBeforeSend{Backend: real}
|
|
grant := g.Grant{Epoch: 1, Owner: g.Digest([]byte(t.Name())), Plan: g.Digest([]byte("plan"))}
|
|
gateway := &g.Gateway{Pool: pool, Scope: real.Bucket, Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20}
|
|
if e := gateway.InstallPlan(ctx, planFor(grant, request(t.Name(), grant))); e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
if e := gateway.Advance(ctx, 0, grant); e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
req := request(t.Name(), grant)
|
|
body := []byte("gateway fixture " + t.Name())
|
|
if _, e := gateway.Execute(ctx, req, body); !errors.Is(e, g.ErrHeld) {
|
|
t.Fatal(e)
|
|
}
|
|
restarted := *gateway
|
|
for i := 0; i < 2; i++ {
|
|
if _, e := restarted.Execute(ctx, req, body); !errors.Is(e, g.ErrHeld) {
|
|
t.Fatal("absent version treated as terminal", e)
|
|
}
|
|
}
|
|
if backend.calls != 1 {
|
|
t.Fatal("PUT replayed", backend.calls)
|
|
}
|
|
next := grant
|
|
next.Epoch++
|
|
if e := restarted.Advance(ctx, 1, next); !errors.Is(e, g.ErrHeld) {
|
|
t.Fatal("unknown operation allowed epoch handover", e)
|
|
}
|
|
var n int
|
|
if pool.QueryRow(ctx, "SELECT count(*) FROM kg_object_gateway_receipts WHERE scope=$1", gateway.Scope).Scan(&n) != nil || n != 0 {
|
|
t.Fatal("absence wrote false terminal receipt")
|
|
}
|
|
}
|
|
func TestGatewayReadOnlyPersistenceBlocksSend(t *testing.T) {
|
|
ctx, pool, backend := fixture(t)
|
|
cfg := pool.Config()
|
|
cfg.ConnConfig.RuntimeParams["default_transaction_read_only"] = "on"
|
|
other, e := pgxpool.NewWithConfig(ctx, cfg)
|
|
if e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
defer other.Close()
|
|
gateway := &g.Gateway{Pool: other, Scope: backend.Bucket, Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20}
|
|
if e = gateway.Ready(ctx); !errors.Is(e, g.ErrPersistence) {
|
|
t.Fatal("read-only persistence accepted", e)
|
|
}
|
|
}
|