fix(objectgateway): 取消整桶独占前置并按维护停写更新
build / build (push) Failing after 2m29s

现象:共享桶部署下网关启动即报 gateway_auditor_key_required,残留过期
审计文件则每次 Execute 报 gateway_exclusive_writer_evidence_unverified;
用户已改为“停目标写入容器→更新→启动”的维护式更新。

根因:Gateway.Ready 强制 Isolation.Verify,main 强制 Ed25519 公钥,等于
把整桶独占证明当作发布前置;这与共享桶、沿用原 AK 的实际部署矛盾。

取舍:删除审计校验与相关 env 依赖,保留 journal、治理 PG 只读 authority、
plan/owner/attempt/epoch 绑定、intent-before-send、单次写与精确版本回滚。
create 仅在精确版本读回且仍为 current 时提交;发现外部 latest 只返回冲突,
不在 confirm 阶段删除(journal create intent 已提交,删除会变成无 delete
intent/receipt 的未记账副作用并让 intent 永久 held)。同 key 不自动合并。

证据:真实二进制无审计/残留审计两环境通过;kg-pgtest-writeset 一次性库 +
随机 MinIO 桶跑 14 个网关用例(含共享桶共存写者、同 key 既有对象拒绝、
精确回滚)全过;umask 022 下 make verify 本地必需项全 success。
This commit is contained in:
姚章浩
2026-11-04 09:45:47 +08:00
parent 017765763a
commit a55319d972
25 changed files with 510 additions and 171 deletions
+27
View File
@@ -1,5 +1,32 @@
# Changelog
## 开发快照(共享桶维护式更新:取消整桶独占前置)
### Changed
- **对象网关不再要求整桶独占证明**:`Gateway.Ready` 只核 journal、PG 持久化设置与后端版本能力,
删除 `Isolation`/`AuditVerifier` 与 `OBJECT_GATEWAY_AUDIT_FILE`、`OBJECT_GATEWAY_AUDIT_PUBLIC_KEY`、
`OBJECT_GATEWAY_ID` 的启动与逐请求校验;旧部署残留值被忽略,不再阻塞默认共享桶路径。治理 PG
SELECT-only authority、plan/owner/attempt/epoch 绑定、intent-before-send、单次写与精确回滚不变。
- **目标 KB 更新走维护窗口**:停目标测试 WeKnora 写入应用容器并读回确认 → 备份本次目标 KB/对象范围
→ 用原 plan/WriteSet 更新 → 启动原容器 → 健康与页面验收;失败按原备份恢复。不再要求排空整桶
全部 writer、RAM 权限清单或独立只读 OSS AK,现有 AK 继续用于 PUT/GET/DELETE/列版本。
### Changed
- **对象网关不再要求整桶独占证明**:启动只核 journal、PG 持久化设置与后端版本能力,旧部署残留的
`OBJECT_GATEWAY_ID`、`OBJECT_GATEWAY_AUDIT_FILE`、`OBJECT_GATEWAY_AUDIT_PUBLIC_KEY` 被忽略,不再阻塞共享桶路径。
治理 PG SELECT-only authority、plan/owner/attempt/epoch 绑定、intent-before-send、单次写与精确回滚不变。
- **更新方式改为维护停写**:停目标测试 WeKnora 写入应用容器并读回确认 → 既有维护/预备份链记录目标 KB 范围 →
原 plan/WriteSet 经正式 Apply 更新 → 启动原容器 → 健康与页面验收;失败走既有恢复链回放原预备份。
不再要求排空整桶全部 writer、RAM 权限清单或独立只读 OSS AK,现有 AK 继续用于 PUT/GET/DELETE/列版本。
### Fixed
- **共享桶不再因同 key 外部写者误报成功**:create 后除精确版本读回外核对当前 latest;并发写者产生更新的当前版本时,
明确返回冲突并保留双方版本,不做任何未记账的删除,也不会在 Reconcile 里把已被外部写者接管的 key 报成 committed。
版本化存储的 PUT 不是原子 create-if-absent,同 key 并发不自动合并,需按回执人工处置。
## 开发快照(B1-R1 / B2-R1 / B5-R1 真实链收口)
### Fixed
+4 -7
View File
@@ -2,8 +2,6 @@ package main
import (
"context"
"crypto/ed25519"
"encoding/base64"
"errors"
"log"
"net"
@@ -35,10 +33,6 @@ func run() error {
if len(data) < 32 || len(control) < 32 || data == control {
return errors.New("gateway_distinct_tokens_required")
}
pub, e := base64.StdEncoding.DecodeString(os.Getenv("OBJECT_GATEWAY_AUDIT_PUBLIC_KEY"))
if e != nil || len(pub) != ed25519.PublicKeySize {
return errors.New("gateway_auditor_key_required")
}
provider, endpoint, bucket := os.Getenv("OBJECT_GATEWAY_PROVIDER"), os.Getenv("OBJECT_GATEWAY_ENDPOINT"), os.Getenv("OBJECT_GATEWAY_BUCKET")
scope := objectgateway.Digest([]byte(provider + "\x00" + endpoint + "\x00" + bucket))
pool, e := pgxpool.New(ctx, os.Getenv("OBJECT_GATEWAY_DATABASE_URL"))
@@ -50,7 +44,10 @@ func run() error {
if e != nil {
return errors.New("gateway_backend_configuration_invalid")
}
g := &objectgateway.Gateway{Pool: pool, Scope: scope, Backend: backend, MaxBytes: 256 * 1024 * 1024, Isolation: objectgateway.AuditVerifier{Path: os.Getenv("OBJECT_GATEWAY_AUDIT_FILE"), PublicKey: pub, Scope: scope, GatewayID: os.Getenv("OBJECT_GATEWAY_ID")}}
// The target bucket is shared with other writers. Admission is the journal
// plus backend capability; retired attestation env vars are deliberately
// ignored so an old deployment file cannot gate this path.
g := &objectgateway.Gateway{Pool: pool, Scope: scope, Backend: backend, MaxBytes: 256 * 1024 * 1024}
if governanceURL := os.Getenv("OBJECT_GATEWAY_GOVERNANCE_DATABASE_URL"); governanceURL != "" {
cfg, e := pgxpool.ParseConfig(governanceURL)
if e != nil {
+82
View File
@@ -0,0 +1,82 @@
package main
import (
"context"
"encoding/base64"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
"time"
)
// startErrorMessage builds the addressed binary and runs it long enough to
// observe why the real entry point refused to listen.
func startErrorMessage(t *testing.T, extra ...string) (string, int) {
t.Helper()
bin := filepath.Join(t.TempDir(), "object-gateway")
build := exec.Command("go", "build", "-o", bin, ".")
if out, err := build.CombinedOutput(); err != nil {
t.Fatalf("build gateway: %v\n%s", err, out)
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
cmd := exec.CommandContext(ctx, bin)
cmd.Env = append([]string{
"PATH=" + os.Getenv("PATH"),
"HOME=" + os.Getenv("HOME"),
"OBJECT_GATEWAY_LISTEN=127.0.0.1:0",
"OBJECT_GATEWAY_DATABASE_URL=postgres://gateway:gateway@127.0.0.1:1/gateway?sslmode=disable&connect_timeout=2",
"OBJECT_GATEWAY_PROVIDER=minio",
"OBJECT_GATEWAY_ENDPOINT=http://127.0.0.1:19100",
"OBJECT_GATEWAY_BUCKET=maintenance-fixture-bucket",
"OBJECT_GATEWAY_REGION=us-east-1",
"OBJECT_GATEWAY_ACCESS_KEY=fixture-access-key",
"OBJECT_GATEWAY_SECRET_KEY=fixture-secret-key",
"OBJECT_GATEWAY_DATA_TOKEN=" + strings.Repeat("d", 32),
"OBJECT_GATEWAY_CONTROL_TOKEN=" + strings.Repeat("c", 32),
}, extra...)
out, err := cmd.CombinedOutput()
if ctx.Err() != nil {
t.Fatalf("gateway did not terminate; output=%s", out)
}
code := 0
if exit, ok := err.(*exec.ExitError); ok {
code = exit.ExitCode()
} else if err != nil {
t.Fatalf("run gateway: %v", err)
}
return string(out), code
}
// The default shared-bucket path must not require a whole-bucket exclusive
// writer attestation. Only the journal/backend admission gate may refuse it,
// and a stale attestation file left behind by an older deployment must not
// gate the default path either.
func TestGatewayEntrypointDoesNotRequireBucketAttestation(t *testing.T) {
for _, tc := range []struct {
name string
env []string
}{
{name: "no_audit_configured"},
{name: "stale_audit_file_ignored", env: []string{
"OBJECT_GATEWAY_ID=retired-gateway",
"OBJECT_GATEWAY_AUDIT_FILE=/nonexistent/expired-attestation.json",
"OBJECT_GATEWAY_AUDIT_PUBLIC_KEY=" + base64.StdEncoding.EncodeToString(make([]byte, 32)),
}},
} {
t.Run(tc.name, func(t *testing.T) {
out, code := startErrorMessage(t, tc.env...)
if code == 0 {
t.Fatalf("gateway exited 0 instead of reporting its admission state: %s", out)
}
if strings.Contains(out, "gateway_auditor_key_required") || strings.Contains(out, "gateway_exclusive_writer_evidence_unverified") {
t.Fatalf("shared-bucket admission still gated by whole-bucket attestation: %s", out)
}
if !strings.Contains(out, "gateway_persistence_unavailable") && !strings.Contains(out, "gateway_backend_not_ready") {
t.Fatalf("gateway did not reach journal/backend admission: %s", out)
}
})
}
}
+2 -2
View File
@@ -1,6 +1,6 @@
# v0.7.1 整库发布的数据契约与覆盖边界
这是候选代码的契约,不是部署许可。真实隔离执行证据在交付目录 `data-contract-r10.md`、`test-summary-r10.json` 及其 raw 索引。PG/ES/MinIO 均为本机隔离真实服务;发布/恢复测试直接调用适配器;B22 另走实际 `librarystack.OpenProductionStack` 和 HTTP 网关。网关的独占审核是测试夹具,不代表生产已隔离;B22 控制读取角色是真实 SELECT-only PG 角色。
这是候选代码的契约,不是部署许可。真实隔离执行证据在交付目录 `data-contract-r10.md`、`test-summary-r10.json` 及其 raw 索引。PG/ES/MinIO 均为本机隔离真实服务;发布/恢复测试直接调用适配器;B22 另走实际 `librarystack.OpenProductionStack` 和 HTTP 网关。网关按限定 key 工作,不要求整桶独占审核;B22 控制读取角色是真实 SELECT-only PG 角色。
## 1. 源 schema、历史身份与准入
@@ -55,7 +55,7 @@
- 真实v071历史catalog镜像对照及UUID/自然键矩阵在r11执行,隔离结论与外部当前目录采样分开。A04已识别嵌套引用有有限支持,旧外部cursor仍不自动转换,不因内部结构正例而升级整项。
- r8/r9进程恢复历史证据只在原发送协议、不可变计划/receipt及相应场景前提不变时复用。本轮对schema/引用准入有收紧,不能把历史执行日志称为当前源码正式进程全链重跑;完整当前隔离适配器回归单列。
- clean-finalization unknown保持held是安全行为,但未满足全部自动收敛要求。若要求自动完成,需要能绑定原写入、完整内容和代次的充分服务端证据;相同干净正文、人工猜测归属或放宽共享规则均不可替代。
- 最终提交/push、远端CI、父复核和本机部署为明确下一门;整体安全验收未齐不部署。生产独占/ACL网络凭据、网关OSS业务端到端与宿主重启/断电是另行授权的部署/破坏性验收,不能作为停止本机授权开发的借口。
- 最终提交/push、远端CI、父复核和本机部署为明确下一门;整体安全验收未齐不部署。网关真实 OSS 业务端到端与宿主重启/断电是另行授权的部署/破坏性验收,不能作为停止本机授权开发的借口。
## 6. 目录与唯一键的有限支持矩阵
+24 -12
View File
@@ -4,12 +4,14 @@
`cmd/object-gateway` 提供 `kg-object-gateway-v1`,只监听 loopback;外部接入需要另行审核的 TLS 终结与网络隔离。治理 PG 保存不可变计划、grant、intent、逐版本 receipt 和 attempt→grant delivery。部署迁移身份显式应用 `internal/objectgateway/schema.sql`;启动不自动迁移。运行身份只授必要 SELECT/INSERT 和 scope 行 UPDATE,不授 DDL/TRUNCATE/证据修改或业务库写权限。
scope 是 provider、原 endpoint、bucket 的 NUL 分隔 SHA256。同一物理桶只能有一个受审核 scope 和一套共享 journal,不能用 endpoint 别名或另一日志库建立平行写域。
scope 是 provider、原 endpoint、bucket 的 NUL 分隔 SHA256。同一物理桶在本网关内只使用一个 scope 和一套共享 journal,不能用 endpoint 别名或另一日志库建立平行写域。这不要求整桶只有本网关一个写者:别的服务可以在同一桶的其它 key 上正常读写。
数据认证端点:`POST /v1/acquire`、`/v1/resolve`、`/v1/execute`、`/v1/reconcile`、`/v1/observe`,以及 `GET /v1/capabilities`。所有请求核验协议和 scope。`capabilities.automatic_grants` 必须为真,正式 stack 才接受自动授权能力;未接控制源、只读角色权限异常、协议不符或不可达均拒绝,无直写回退。
PG 行锁负责原子准入,不是远端存储 fence。intent 必须持久提交后才发送一次请求;连接死亡、租约过期不清除未知 intent。任何未终态 intent 均阻止 scope 换 epoch。重复 operation 只读回执或精确对账,不重发 PUT;暂时查不到版本仍 held。删除必须绑定已确认 create 的精确 VersionID,不删 latest、不生成删除标记。
同 key 竞争按 key 判定,不按桶判定:create 前已存在任何版本 → 回执 `rejected` 且不修改该对象;PUT 后除精确版本读回外还核对当前 latest,若并发写者已产生更新的当前版本,则只按精确 VersionID 撤回本次写入,保留对方版本并返回冲突;读不到当前版本时保持 held,不删除、不猜测。版本化存储的 PUT 不是原子 create-if-absent,因此“本 key 多数情况下归本计划”不是存储层保证。
不可变证据表拒绝 UPDATE/DELETE/TRUNCATE。Ready 核 fsync、synchronous_commit、full_page_writes、主库可写及必要表/触发器;管理员与迁移身份仍是可信边界,触发器不能抵御 DBA 的 DDL。
## 正式控制交付路径
@@ -54,13 +56,13 @@ SELECT-only治理角色需要读取`library_jobs`、`library_operations`、`libr
- `OBJECT_GATEWAY_ARCHIVE_ROOT`:原持久计划所在归档绝对路径,部署只读挂载。
- `OBJECT_GATEWAY_TARGET_ID`:与正式 worker `LibraryTargetIdentity()` 完全相同的逻辑目标身份;不能用另一个环境的值。
- `OBJECT_GATEWAY_PROVIDER`、`OBJECT_GATEWAY_ENDPOINT`、`OBJECT_GATEWAY_BUCKET`、`OBJECT_GATEWAY_REGION`:精确存储身份。
- `OBJECT_GATEWAY_ACCESS_KEY`、`OBJECT_GATEWAY_SECRET_KEY`:仅网关持有的对象写凭据。
- `OBJECT_GATEWAY_ACCESS_KEY`、`OBJECT_GATEWAY_SECRET_KEY`:网关持有的对象凭据,需在目标桶 PUT/GET/按版本 DELETE/列举版本。沿用现有 AK 即可,不要求另建独立只读或独占写凭据。
- `OBJECT_GATEWAY_DATA_TOKEN`、`OBJECT_GATEWAY_CONTROL_TOKEN`:不同且至少 32 字节;普通 worker 仅持有 data token。
- `OBJECT_GATEWAY_ID`、`OBJECT_GATEWAY_AUDIT_FILE`、`OBJECT_GATEWAY_AUDIT_PUBLIC_KEY`:独立审核者的 Ed25519 签名隔离证据,绑定协议、scope、gateway ID、时间窗及 writer/policy/拒写/排空摘要。审核私钥不能交给 worker 或网关。
- `OBJECT_GATEWAY_ID`、`OBJECT_GATEWAY_AUDIT_FILE`、`OBJECT_GATEWAY_AUDIT_PUBLIC_KEY`:**已退役**。旧部署残留的值被忽略,不参与启动、Ready、每次 Execute 或能力判断;不需要管理员签发整桶独占证明,也不会因过期文件拒绝发布。
秘密只经服务环境/私有文件注入,不进 argv、URL 或日志。所有调用端错误都使用固定原因码。
正式 worker 使用 `LIBRARY_OBJECT_GATEWAY_URL` 和 `LIBRARY_OBJECT_GATEWAY_TOKEN`;不再依赖预先生成的 grant 文件。`LIBRARY_OBJECT_GATEWAY_GRANT_DIRECTORY` 保留旧离线兼容用途,不是自动模式的授权来源。发布 stack 对所有对象 provider 强制网关;S3 构造仍用于只读捕获/预备份,部署必须使用独立只读凭据。
正式 worker 使用 `LIBRARY_OBJECT_GATEWAY_URL` 和 `LIBRARY_OBJECT_GATEWAY_TOKEN`;不再依赖预先生成的 grant 文件。`LIBRARY_OBJECT_GATEWAY_GRANT_DIRECTORY` 保留旧离线兼容用途,不是自动模式的授权来源。发布 stack 对所有对象 provider 强制网关;S3 构造仍用于只读捕获/预备份,可继续使用现有凭据,不再要求独立只读 OSS AK。
## 已关闭的本仓旁路
@@ -70,24 +72,34 @@ SELECT-only治理角色需要读取`library_jobs`、`library_operations`、`libr
- `librarystack` 不再为缺网关的 MinIO/S3 返回直写 writer。
- `libraryctl` capture 的 S3 构造与读取保留;未把读取当作写入口禁用。
这不等于已经撤销运行环境旧进程的凭据;源码变更尚未部署。`librarydrill` 等无 owned-attempt 的旧写型工具不能靠静态 grant 绕过自动模式,需补正式执行身份后才可用于写入。低层 S3 adapter 仍用于网关后端、读取及显式隔离夹具;部署权限才负责防止外部旧进程直接调用存储。
这不等于已经撤销运行环境旧进程的凭据;源码变更尚未部署。`librarydrill` 等无 owned-attempt 的旧写型工具不能靠静态 grant 绕过自动模式,需补正式执行身份后才可用于写入。低层 S3 adapter 仍用于网关后端、读取及显式隔离夹具;共享桶不要求撤除其它写者,跨写者的同 key 竞争按冲突/held 处理。
## 实测与未完成边界
已实跑:真实治理 PG(独立 SELECT-only 登录角色)、原 archive、数据 HTTP、业务 PG 与 MinIO 的自动授权/跨 epoch 链;未锚定计划在第一笔 PG 前拒绝,接管保持原发送身份且不增加 PUT。另有真实网关二进制的两次中断验证。隔离审计 authority/故障注入仍是测试设施,不是生产权限证明。
已实跑:真实治理 PG(独立 SELECT-only 登录角色)、原 archive、数据 HTTP、业务 PG 与 MinIO 的自动授权/跨 epoch 链;未锚定计划在第一笔 PG 前拒绝,接管保持原发送身份且不增加 PUT。另有真实网关二进制的两次中断验证,以及共享桶下“无整桶证明启动 + 其它 key 并发写入 + 同 key 既有对象拒绝”的真实 PG/MinIO 用例。历史隔离审计 authority/故障注入是当时的测试设施,不是当前部署前置。
正式 worker 隔离进程证据:真实 Docker 入口闸门/队列、只读 worker 对象身份、独立治理读取角色、网关与业务 PG/ES/MinIO;首次 PG 后和对象提交后两次 SIGKILL,同一原 job/operation/plan 的三代 attempt 完成发布,原对象发送 epoch 与回执不变。隔离构建标签 `kg_isolated_drill` 仅把真实租约缩短至4秒,并拒绝非 loopback 一次性 r8 库;正常构建仍为10分钟。维护上游是无业务凭据的独立 HTTP 夹具,不是运行中的 WeKnora;签名证据绑定本次真实拒写/排空材料,不代表生产独占审核。
已补独立 `library.recover` 正式worker链、恢复中再次SIGKILL、完整before-image/CAS/精确版本读回,以及来源/目标/原job/计划/operation/任意队列身份拒绝。ES故障代理只转发真实ES并丢弃已提交响应,未伪造服务响应。
尚未完成:clean-finalization unknown的充分归属与自动收敛;独立运行环境只读对象凭据、ACL/网络和全部writer排空;网关真实OSS端到端;原D0/F11/F12与55项其余缺口。测试/隔离验收与真实部署独占分列,不把生产变更授权当成本机开发授权的前置。
尚未完成:clean-finalization unknown的充分归属与自动收敛;网关真实 OSS 端到端与真实部署验收(父执行);原D0/F11/F12与55项其余缺口。测试/隔离验收与真实部署分列。
维护隔离、独占审核与 grant 是独立前置。配置 true、健康 200、测试自签 authority 都不能代替已部署拒写/排空证据。目前未签发真实环境独占证明,未改生产权限或上游 WeKnora 源码/Schema。
整桶独占审核、全部 writer 排空、独立只读对象凭据已不再是发布前置;维护窗口内停目标写入容器并读回确认即可更新。网关按 key 拒绝冲突、精确版本回滚,但不承诺无关写者同时修改同一 key 时仍能自动完成——那种情况会明确冲突或 held,需要人工按回执处理。未改生产权限或上游 WeKnora 源码/Schema。
## 维护式更新(已采用)
目标测试 KB 的更新走维护窗口,不再要求证明整桶只有一个写者:
1. 停目标测试 WeKnora 的写入应用容器(只停该服务,不停机器人、适配器、共享 PG/ES 或其它 KB 的服务),并由容器状态/健康读回确认已停止。这是本轮文档化的运维步骤;代码侧没有新增停容器实现。
2. 用既有维护/预备份链记录本次目标 KB/对象范围:实例级维护窗口与写者停写证据由 `internal/libraryfreeze` + `internal/librarymaintenance` 采集,`libraryctl capture`(页面入口 `POST /api/v1/library/captures`)完成闸门排空→停写者→导出快照并保留 `post_stop_digest`;不把通用主机备份当作 KB 预备份的替代。
3. 用原 plan、原 WriteSet 摘要和限定对象清单走既有 `ProductionLibraryTarget.Apply`(经正式 worker owned attempt 与网关 grant)更新资源;网关沿用现有 AK,不换凭据、不签独占证明。
4. 启动原容器,健康检查与页面业务验收;失败走既有 `library.recover`/`RestoreDrill` 恢复链按原预备份回放并启动原服务,不无限留停机。
5. 无关服务与无关 key 全程不需要停;同 key 若仍有外部写者,按上节的冲突/held 处理:双方版本都保留,由人按证据处置,不自动合并。
部署需变更的组件:`object-gateway`(本候选镜像)、`worker`(同提交镜像),以及两者 env 中移除/留空已退役的 `OBJECT_GATEWAY_ID`、`OBJECT_GATEWAY_AUDIT_FILE`、`OBJECT_GATEWAY_AUDIT_PUBLIC_KEY`。`api`、`web` 无协议变更,可不同步。journal、治理 PG 只读 authority、archive 与 target identity 配置保持不变。
## 部署与回退(未执行)
未来测试主机 `120.76.217.51`、生产 `120.76.218.137` 各自独立批准。每批先冻结主机、桶、全部角色/STS/预签名写入口、端口/网络规则、影响、维护窗口与回滚;当前没有执行这些外部变更。
测试主机 `120.76.217.51`、生产 `120.76.218.137` 各自独立批准;本候选未在任一主机执行停机、备份、更新或启动。r5 的已完成 OSS 测试授权不能重复使用。
先关新写准入→排空并对账旧请求→撤旧直写权限/切只读身份并实测拒写→独立审核→批准的新限定对象验收→父复核,才可切业务客户端。r5 的已完成 OSS 测试授权不能重复使用。
回退先关准入并保留 journal;所有已发送请求有终态才可撤网关,unknown 继续 held。不得因网关故障自动恢复旧直写凭据。权限/网络回退须另批。当前整模块 INCOMPLETE,禁止部署部分候选。
按“停目标写入容器→读回确认→备份目标 KB 范围→更新资源→启动→健康/页面验收”执行;失败按原备份恢复并启动原服务。回退先关新写准入并保留 journal;所有已发送请求有终态才可撤网关,unknown 继续 held。不得因网关故障自动恢复旧直写凭据。当前整模块仍未完成真实 OSS 端到端与 clean-finalization 收敛,禁止把本候选当作完整生产就绪。
@@ -24,7 +24,7 @@ func TestGatewayAcquisitionPersistsAttemptDelivery(t *testing.T) {
expected := g.Grant{Epoch: 1, Owner: g.Digest([]byte(claim.JobID)), Plan: claim.PlanDigest}
req := request("acquired-object", expected)
authority := &currentPlanAuthority{claim, planFor(expected, req)}
gateway := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20, Authority: authority}
gateway := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, MaxBytes: 1 << 20, Authority: authority}
grant, e := gateway.Acquire(ctx, claim)
if e != nil || grant != expected {
t.Fatal("anchored active attempt grant unavailable", e)
@@ -79,7 +79,7 @@ func TestGatewayAcquisitionCannotHandOverUnknownSend(t *testing.T) {
req := request("acquire-held", expected)
body := []byte("gateway fixture acquire-held")
authority := &currentPlanAuthority{claim, planFor(expected, req)}
gateway := &g.Gateway{Pool: pool, Scope: real.Bucket, Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20, Authority: authority}
gateway := &g.Gateway{Pool: pool, Scope: real.Bucket, Backend: backend, MaxBytes: 1 << 20, Authority: authority}
if _, e := gateway.Acquire(ctx, claim); e != nil {
t.Fatal(e)
}
@@ -0,0 +1,26 @@
package objectgateway
import (
"encoding/json"
"time"
)
// Attestation is the retired whole-bucket exclusive-writer evidence format.
// It is kept only so historical audit artifacts remain readable; no runtime
// path loads, verifies or requires it. Admission is the journal, the backend
// capability and the frozen plan, all of which work on a shared bucket.
type Attestation struct {
Protocol string `json:"protocol"`
Scope string `json:"scope"`
GatewayID string `json:"gateway_id"`
IssuedAt time.Time `json:"issued_at"`
ExpiresAt time.Time `json:"expires_at"`
WriterInventory string `json:"writer_inventory_sha256"`
PolicyEvidence string `json:"policy_evidence_sha256"`
DenialEvidence string `json:"denial_evidence_sha256"`
DrainEvidence string `json:"drain_evidence_sha256"`
}
type SignedAttestation struct {
Payload json.RawMessage `json:"payload"`
Signature []byte `json:"signature"`
}
-57
View File
@@ -1,57 +0,0 @@
package objectgateway
import (
"context"
"crypto/ed25519"
"encoding/json"
"os"
"time"
)
// Attestation is issued by an independent deployment auditor, never by a worker.
// Its signature authenticates an audit, not the truth of self-reported config.
// Runtime cutover must invalidate/replace the audit whenever writer rights change.
type Attestation struct {
Protocol string `json:"protocol"`
Scope string `json:"scope"`
GatewayID string `json:"gateway_id"`
IssuedAt time.Time `json:"issued_at"`
ExpiresAt time.Time `json:"expires_at"`
WriterInventory string `json:"writer_inventory_sha256"`
PolicyEvidence string `json:"policy_evidence_sha256"`
DenialEvidence string `json:"denial_evidence_sha256"`
DrainEvidence string `json:"drain_evidence_sha256"`
}
type SignedAttestation struct {
Payload json.RawMessage `json:"payload"`
Signature []byte `json:"signature"`
}
type AuditVerifier struct {
Path string
PublicKey ed25519.PublicKey
Scope, GatewayID string
}
func (v AuditVerifier) Verify(ctx context.Context) error {
if ctx.Err() != nil {
return ErrIsolation
}
st, e := os.Lstat(v.Path)
if e != nil || !st.Mode().IsRegular() || st.Size() > 16384 || st.Mode().Perm()&0022 != 0 {
return ErrIsolation
}
raw, e := os.ReadFile(v.Path)
if e != nil {
return ErrIsolation
}
var signed SignedAttestation
var a Attestation
if json.Unmarshal(raw, &signed) != nil || len(v.PublicKey) != ed25519.PublicKeySize || !ed25519.Verify(v.PublicKey, signed.Payload, signed.Signature) || json.Unmarshal(signed.Payload, &a) != nil {
return ErrIsolation
}
now := time.Now()
if a.Protocol != Protocol || a.Scope != v.Scope || a.GatewayID != v.GatewayID || a.GatewayID == "" || a.IssuedAt.After(now) || !a.ExpiresAt.After(now) || a.ExpiresAt.Sub(a.IssuedAt) > time.Hour || !validHash(a.WriterInventory) || !validHash(a.PolicyEvidence) || !validHash(a.DenialEvidence) || !validHash(a.DrainEvidence) {
return ErrIsolation
}
return nil
}
-59
View File
@@ -1,59 +0,0 @@
package objectgateway_test
import (
"context"
"crypto/ed25519"
"crypto/rand"
"encoding/json"
g "knowledge-governance/internal/objectgateway"
"os"
"path/filepath"
"testing"
"time"
)
func TestGatewayAuditCannotBeConfigurationBoolean(t *testing.T) {
pub, priv, e := ed25519.GenerateKey(rand.Reader)
if e != nil {
t.Fatal(e)
}
dir := t.TempDir()
path := filepath.Join(dir, "audit.json")
verifier := g.AuditVerifier{Path: path, PublicKey: pub, Scope: "target", GatewayID: "gateway"}
if verifier.Verify(context.Background()) == nil {
t.Fatal("missing audit accepted")
}
if os.WriteFile(path, []byte(`{"isolated":true,"ready":true}`), 0600) != nil {
t.Fatal("fixture")
}
if verifier.Verify(context.Background()) == nil {
t.Fatal("boolean accepted as isolation proof")
}
a := g.Attestation{Protocol: g.Protocol, Scope: "target", GatewayID: "gateway", IssuedAt: time.Now().Add(-time.Minute), ExpiresAt: time.Now().Add(30 * time.Minute), WriterInventory: g.Digest([]byte("inventory")), PolicyEvidence: g.Digest([]byte("policy")), DenialEvidence: g.Digest([]byte("denials")), DrainEvidence: g.Digest([]byte("drain"))}
write := func(v g.Attestation) {
t.Helper()
raw, _ := json.Marshal(v)
signed, _ := json.Marshal(g.SignedAttestation{Payload: raw, Signature: ed25519.Sign(priv, raw)})
if os.WriteFile(path, signed, 0600) != nil {
t.Fatal("fixture")
}
}
write(a)
if verifier.Verify(context.Background()) != nil {
t.Fatal("valid test auditor signature rejected")
}
for _, mutate := range []func(*g.Attestation){func(v *g.Attestation) { v.Scope = "other" }, func(v *g.Attestation) { v.GatewayID = "other" }, func(v *g.Attestation) { v.ExpiresAt = time.Now().Add(-time.Second) }, func(v *g.Attestation) { v.DenialEvidence = "" }, func(v *g.Attestation) { v.IssuedAt = time.Now().Add(time.Hour) }, func(v *g.Attestation) { v.ExpiresAt = time.Now().Add(24 * time.Hour) }} {
bad := a
mutate(&bad)
write(bad)
if verifier.Verify(context.Background()) == nil {
t.Fatal("invalid isolation audit accepted")
}
}
write(a)
other, _, _ := ed25519.GenerateKey(rand.Reader)
verifier.PublicKey = other
if verifier.Verify(context.Background()) == nil {
t.Fatal("untrusted signer accepted")
}
}
+43
View File
@@ -0,0 +1,43 @@
package objectgateway
import "context"
// sendCreate performs exactly one non-retried versioned PUT and then confirms
// the result. Versioned object stores do not provide atomic create-if-absent,
// so the confirmation re-reads the current version to make sure this request's
// version is still the one a reader would observe.
func sendCreate(ctx context.Context, b Backend, r Request, body []byte) (Version, error) {
v, err := b.Put(ctx, r.Key, body, r.token())
if err != nil {
// Unknown outcome: the intent stays unresolved, nothing is retried and
// nothing is deleted because no version id is confirmed.
return Version{}, ErrHeld
}
return confirmCreate(ctx, b, r, v)
}
// confirmCreate binds the receipt to a version that is both the exact version
// this request wrote and the current version of the key. It is read-only: the
// create intent is already committed, so withdrawing a version here would be an
// unrecorded mutation with no delete intent or receipt and would leave the
// create intent permanently held. A concurrent writer that produced a newer
// current version therefore wins by explicit conflict; both versions stay in
// the store and an operator resolves the key with evidence.
func confirmCreate(ctx context.Context, b Backend, r Request, v Version) (Version, error) {
exact, exists, err := b.Observe(ctx, r.Key, v.ID)
if err != nil || !exists || !r.matches(exact) || v != exact {
return Version{}, ErrHeld
}
latest, current, err := b.Observe(ctx, r.Key, "")
if err != nil || !current {
// Cannot tell whether our write is still current: keep the intent held
// rather than mutating anything on a guess.
return Version{}, ErrHeld
}
if latest.ID != v.ID {
// Another writer owns the current version. Report the conflict and leave
// both versions untouched; this path never merges them automatically.
return Version{}, ErrConflict
}
return v, nil
}
+153
View File
@@ -0,0 +1,153 @@
package objectgateway
import (
"context"
"errors"
"testing"
)
type fakeBackend struct {
byID map[string]Version
latest Version
latestExists bool
latestErr error
putVersion Version
putErr error
putCalls int
deletes []string
deleteErr error
}
func (f *fakeBackend) Ready(context.Context) error { return nil }
func (f *fakeBackend) Observe(_ context.Context, _ string, id string) (Version, bool, error) {
if id == "" {
if f.latestErr != nil {
return Version{}, false, f.latestErr
}
return f.latest, f.latestExists, nil
}
v, ok := f.byID[id]
return v, ok, nil
}
func (f *fakeBackend) List(context.Context, string) ([]Version, error) { return nil, nil }
func (f *fakeBackend) Put(context.Context, string, []byte, string) (Version, error) {
f.putCalls++
if f.putErr != nil {
return Version{}, f.putErr
}
return f.putVersion, nil
}
func (f *fakeBackend) Delete(_ context.Context, _ string, v Version) error {
if f.deleteErr != nil {
return f.deleteErr
}
f.deletes = append(f.deletes, v.ID)
delete(f.byID, v.ID)
return nil
}
func createFixture(t *testing.T) (Request, Version) {
t.Helper()
body := []byte("shared bucket object")
grant := Grant{Epoch: 1, Owner: Digest([]byte("owner")), Plan: Digest([]byte("plan"))}
r := Request{Grant: grant, Operation: Digest([]byte("operation")), Key: "kb/bound/object-1", Action: "create", SHA256: Digest(body), Bytes: int64(len(body))}
v := Version{ID: "version-own", Owner: r.token(), SHA256: r.SHA256, Bytes: r.Bytes, ETag: `"etag-own"`}
return r, v
}
func TestCreateConfirmationDetectsNewerForeignVersionWithoutSideEffects(t *testing.T) {
r, own := createFixture(t)
foreign := Version{ID: "version-foreign", Owner: Digest([]byte("foreign-owner")), SHA256: Digest([]byte("foreign")), Bytes: 7, ETag: `"etag-foreign"`}
f := &fakeBackend{byID: map[string]Version{own.ID: own, foreign.ID: foreign}, latest: foreign, latestExists: true, putVersion: own}
if _, err := confirmCreate(context.Background(), f, r, own); !errors.Is(err, ErrConflict) {
t.Fatalf("concurrent newer version accepted as committed: %v", err)
}
// Confirmation runs after the journal intent exists. Deleting here would be
// an unrecorded mutation with no delete intent/receipt and would strand the
// create intent in held forever, so confirmation must be read-only.
if len(f.deletes) != 0 {
t.Fatalf("confirmation deleted without a delete intent/receipt: %v", f.deletes)
}
if _, ok := f.byID[own.ID]; !ok {
t.Fatal("own version removed without a recorded delete")
}
if _, ok := f.byID[foreign.ID]; !ok {
t.Fatal("foreign version was deleted")
}
}
func TestCreateConfirmationCommitsOnlyWhenOwnVersionIsCurrent(t *testing.T) {
r, own := createFixture(t)
f := &fakeBackend{byID: map[string]Version{own.ID: own}, latest: own, latestExists: true, putVersion: own}
got, err := confirmCreate(context.Background(), f, r, own)
if err != nil || got != own {
t.Fatalf("own current version not committed: %v %v", got, err)
}
if len(f.deletes) != 0 {
t.Fatalf("unexpected rollback %v", f.deletes)
}
}
func TestCreateConfirmationHoldsWithoutDeletingWhenCurrentVersionUnreadable(t *testing.T) {
r, own := createFixture(t)
f := &fakeBackend{byID: map[string]Version{own.ID: own}, putVersion: own, latestErr: errors.New("read failed")}
if _, err := confirmCreate(context.Background(), f, r, own); !errors.Is(err, ErrHeld) {
t.Fatalf("unreadable current version treated as terminal: %v", err)
}
if len(f.deletes) != 0 {
t.Fatalf("deleted on an unknown observation: %v", f.deletes)
}
}
func TestCreateConfirmationHoldsOnExactReadbackMismatch(t *testing.T) {
r, own := createFixture(t)
other := own
other.ETag = `"different"`
f := &fakeBackend{byID: map[string]Version{own.ID: other}, latest: other, latestExists: true, putVersion: own}
if _, err := confirmCreate(context.Background(), f, r, own); !errors.Is(err, ErrHeld) {
t.Fatalf("mismatched readback treated as committed: %v", err)
}
if len(f.deletes) != 0 {
t.Fatalf("deleted on mismatch: %v", f.deletes)
}
}
func TestCreateSendIsSingleAttemptAndNeverDeletesUnknownPut(t *testing.T) {
r, _ := createFixture(t)
f := &fakeBackend{putErr: errors.New("transport lost")}
if _, err := sendCreate(context.Background(), f, r, []byte("shared bucket object")); !errors.Is(err, ErrHeld) {
t.Fatalf("unknown PUT treated as terminal: %v", err)
}
if f.putCalls != 1 {
t.Fatalf("PUT retried: calls=%d", f.putCalls)
}
if len(f.deletes) != 0 {
t.Fatalf("deleted without a confirmed version: %v", f.deletes)
}
}
func TestPlanAuthorizesOnlyFrozenObjectList(t *testing.T) {
body := []byte("planned")
owner, digest := Digest([]byte("owner")), Digest([]byte("plan-digest"))
p := Plan{Digest: digest, Owner: owner, Objects: []ObjectBinding{{Operation: Digest([]byte("op-1")), Key: "kb/allowed/1", SHA256: Digest(body), Bytes: int64(len(body))}}}
allowed := Request{Grant: Grant{1, owner, digest}, Operation: Digest([]byte("op-1")), Key: "kb/allowed/1", Action: "create", SHA256: Digest(body), Bytes: int64(len(body))}
if !p.allows(allowed) {
t.Fatal("frozen object rejected")
}
for name, mutate := range map[string]func(*Request){
"other_kb_key": func(r *Request) { r.Key = "kb/other-kb/1" },
"other_object": func(r *Request) { r.Key = "kb/allowed/2" },
"other_operation": func(r *Request) { r.Operation = Digest([]byte("op-2")) },
} {
got := allowed
mutate(&got)
if p.allows(got) {
t.Fatalf("%s authorized outside the frozen write set", name)
}
}
dup := p
dup.Objects = append(dup.Objects, p.Objects[0])
if dup.valid() {
t.Fatal("duplicate key accepted in plan")
}
}
+17 -10
View File
@@ -20,17 +20,18 @@ type Gateway struct {
Pool *pgxpool.Pool
Scope string
Backend Backend
Isolation IsolationVerifier
MaxBytes int64
}
// Ready gates admission on the local journal and the object backend only. A
// shared bucket is the supported deployment shape: unrelated writers on other
// keys do not need to be drained, and no whole-bucket exclusivity evidence is
// required. Per-key competition is handled at execute time, not by a global
// "only writer" attestation.
func (g *Gateway) Ready(ctx context.Context) error {
if g.Pool == nil || g.Scope == "" || g.Backend == nil || g.MaxBytes < 1 {
return ErrPersistence
}
if g.Isolation == nil || g.Isolation.Verify(ctx) != nil {
return ErrIsolation
}
var sync, fsync, full string
if g.Pool.QueryRow(ctx, "SELECT current_setting('synchronous_commit'), current_setting('fsync'), current_setting('full_page_writes')").Scan(&sync, &fsync, &full) != nil || sync != "on" || fsync != "on" || full != "on" {
return ErrPersistence
@@ -249,13 +250,9 @@ func (g *Gateway) execute(ctx context.Context, r Request, body []byte, authoriza
if exists {
return g.finish(ctx, r, "rejected", Version{})
}
v, err := g.Backend.Put(ctx, r.Key, body, r.token())
v, err := sendCreate(ctx, g.Backend, r, body)
if err != nil {
return Receipt{}, ErrHeld
}
exact, exists, err := g.Backend.Observe(ctx, r.Key, v.ID)
if err != nil || !exists || !r.matches(exact) || v != exact {
return Receipt{}, ErrHeld
return Receipt{}, err
}
return g.finish(ctx, r, "committed", v)
}
@@ -322,5 +319,15 @@ func (g *Gateway) Reconcile(ctx context.Context, r Request) (Receipt, error) {
if err != nil || !exists || exact != found[0] {
return Receipt{}, ErrHeld
}
latest, current, err := g.Backend.Observe(ctx, r.Key, "")
if err != nil || !current {
return Receipt{}, ErrHeld
}
if latest.ID != exact.ID {
// Another writer owns the current version of this key. Do not delete
// anything during reconcile and do not report a committed create for a
// key this plan no longer controls.
return Receipt{}, ErrConflict
}
return g.finish(ctx, r, "committed", exact)
}
@@ -30,9 +30,6 @@ import (
g "knowledge-governance/internal/objectgateway"
)
type verifiedFixture struct{}
func (verifiedFixture) Verify(context.Context) error { return nil }
func fixture(t *testing.T) (context.Context, *pgxpool.Pool, *g.S3Backend) {
t.Helper()
dsn := os.Getenv("KG_GATEWAY_TEST_DATABASE_URL")
@@ -118,7 +115,7 @@ func TestGatewayReadinessRequiresJournal(t *testing.T) {
t.Fatal(e)
}
defer other.Close()
gw := &g.Gateway{Pool: other, Scope: "missing-journal", Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20}
gw := &g.Gateway{Pool: other, Scope: "missing-journal", Backend: backend, MaxBytes: 1 << 20}
if e := gw.Ready(ctx); !errors.Is(e, g.ErrPersistence) {
t.Fatalf("missing durable journal accepted: %v", e)
}
@@ -126,7 +123,7 @@ func TestGatewayReadinessRequiresJournal(t *testing.T) {
func TestGatewayRealReplayAndImmutableLedger(t *testing.T) {
ctx, pool, backend := fixture(t)
grant := g.Grant{Epoch: 1, Owner: g.Digest([]byte(t.Name())), Plan: g.Digest([]byte("fixed-plan"))}
gw := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20}
gw := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, MaxBytes: 1 << 20}
if _, exists, err := backend.Observe(ctx, "missing-key-probe", ""); err != nil || exists {
t.Fatalf("initial authoritative absence unavailable: %v exists=%v", err, exists)
}
-2
View File
@@ -239,8 +239,6 @@ func (c *Client) call(ctx context.Context, path string, in, out any) error {
return ErrHeld
case ErrPersistence.Error():
return ErrPersistence
case ErrIsolation.Error():
return ErrIsolation
}
return errors.New("gateway_request_refused")
}
@@ -14,7 +14,7 @@ import (
func TestGatewayTargetIdentityOnEveryCall(t *testing.T) {
ctx, pool, backend := fixture(t)
grant := g.Grant{Epoch: 1, Owner: g.Digest([]byte(t.Name())), Plan: g.Digest([]byte("fixed-plan"))}
gw := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20}
gw := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, MaxBytes: 1 << 20}
if e := gw.InstallPlan(ctx, planFor(grant, request(t.Name(), grant))); e != nil {
t.Fatal(e)
}
@@ -10,7 +10,7 @@ func TestGatewayIntentCommitFailureNeverSends(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}
gateway := &g.Gateway{Pool: pool, Scope: real.Bucket, Backend: backend, MaxBytes: 1 << 20}
if e := gateway.InstallPlan(ctx, planFor(grant, request(t.Name(), grant))); e != nil {
t.Fatal(e)
}
@@ -9,7 +9,7 @@ import (
func TestGatewayGrantCannotAuthorizeUnplannedKey(t *testing.T) {
ctx, pool, backend := fixture(t)
grant := g.Grant{Epoch: 1, Owner: g.Digest([]byte(t.Name())), Plan: g.Digest([]byte("fixed-plan"))}
gateway := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20}
gateway := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, MaxBytes: 1 << 20}
if e := gateway.Advance(ctx, 0, grant); !errors.Is(e, g.ErrConflict) {
t.Fatal("unregistered plan grant", e)
}
-2
View File
@@ -19,7 +19,6 @@ var (
ErrPersistence = errors.New("gateway_persistence_unavailable")
ErrAuthority = errors.New("gateway_control_authority_unavailable")
ErrAuthorityPrivileges = errors.New("gateway_control_reader_privileges_invalid")
ErrIsolation = errors.New("gateway_exclusive_writer_evidence_unverified")
ErrRejected = errors.New("gateway_create_rejected_existing_object")
)
@@ -60,7 +59,6 @@ type Backend interface {
Put(context.Context, string, []byte, string) (Version, error)
Delete(context.Context, string, Version) error
}
type IsolationVerifier interface{ Verify(context.Context) error }
func Digest(b []byte) string { s := sha256.Sum256(b); return hex.EncodeToString(s[:]) }
func validHash(s string) bool {
@@ -0,0 +1,114 @@
package objectgateway_test
import (
"bytes"
"net"
"os"
"testing"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/s3"
g "knowledge-governance/internal/objectgateway"
)
// A maintenance update of the target KB runs while the same bucket still
// serves other writers. Publishing planned objects must neither require a
// whole-bucket exclusivity attestation nor disturb an unrelated key or a
// pre-existing object at a planned key. External writes here are made before
// execute (coexistence), not interleaved concurrently with the PUT.
func TestGatewaySharedBucketCoexistenceWithOtherWriters(t *testing.T) {
ctx, pool, backend := fixture(t)
endpoint := os.Getenv("KG_GATEWAY_TEST_ENDPOINT")
scope := g.Digest([]byte("minio\x00" + endpoint + "\x00" + backend.Bucket))
listener, e := net.Listen("tcp", "127.0.0.1:0")
if e != nil {
t.Fatal(e)
}
addr := listener.Addr().String()
listener.Close()
data, control := g.Digest([]byte("data "+scope)), g.Digest([]byte("control "+scope))
client, e := g.NewClient("http://"+addr, data, scope)
if e != nil {
t.Fatal(e)
}
admin, _ := g.NewClient("http://"+addr, control, scope)
// Deliberately no OBJECT_GATEWAY_AUDIT_FILE / OBJECT_GATEWAY_AUDIT_PUBLIC_KEY:
// the default shared-bucket path must start without whole-bucket evidence.
env := []string{
"PATH=" + os.Getenv("PATH"),
"HOME=" + os.Getenv("HOME"),
"OBJECT_GATEWAY_LISTEN=" + addr,
"OBJECT_GATEWAY_DATABASE_URL=" + os.Getenv("KG_GATEWAY_TEST_DATABASE_URL"),
"OBJECT_GATEWAY_PROVIDER=minio",
"OBJECT_GATEWAY_ENDPOINT=" + endpoint,
"OBJECT_GATEWAY_BUCKET=" + backend.Bucket,
"OBJECT_GATEWAY_REGION=us-east-1",
"OBJECT_GATEWAY_ACCESS_KEY=" + os.Getenv("KG_GATEWAY_TEST_ACCESS_KEY"),
"OBJECT_GATEWAY_SECRET_KEY=" + os.Getenv("KG_GATEWAY_TEST_SECRET_KEY"),
"OBJECT_GATEWAY_DATA_TOKEN=" + data,
"OBJECT_GATEWAY_CONTROL_TOKEN=" + control,
}
startGateway(t, env, client)
if e = client.ReadyForAcquisition(ctx); e == nil {
t.Fatal("gateway without control authority advertised automatic grants")
}
// Another service keeps writing unrelated keys during the maintenance window.
externalOwner := g.Digest([]byte("other-service"))
unrelatedKey := "other-service/unrelated-" + g.Digest([]byte(backend.Bucket))[:8]
unrelatedBody := []byte("unrelated service body")
if _, e = backend.Client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(backend.Bucket), Key: aws.String(unrelatedKey), Body: bytes.NewReader(unrelatedBody), Metadata: map[string]string{"kg-release-owner": externalOwner}}); e != nil {
t.Fatalf("external unrelated write failed: %v", e)
}
unrelated, _, e := backend.Observe(ctx, unrelatedKey, "")
if e != nil {
t.Fatal(e)
}
// A planned key already holds a foreign object: the plan must refuse it and
// leave that object byte-identical.
existingKey := "unchanged/existing-object"
existingBody := []byte("pre-existing production content")
if _, e = backend.Client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(backend.Bucket), Key: aws.String(existingKey), Body: bytes.NewReader(existingBody), Metadata: map[string]string{"kg-release-owner": externalOwner}}); e != nil {
t.Fatal(e)
}
existing, _, e := backend.Observe(ctx, existingKey, "")
if e != nil {
t.Fatal(e)
}
grant := g.Grant{Epoch: 1, Owner: g.Digest([]byte("maintenance-owner")), Plan: g.Digest([]byte("maintenance-plan"))}
fresh := request("shared-fresh", grant)
clash := request("shared-clash", grant)
clash.Key = existingKey
if e = admin.InstallPlan(ctx, planFor(grant, fresh, clash)); e != nil {
t.Fatal(e)
}
if e = admin.Advance(ctx, 0, grant); e != nil {
t.Fatal(e)
}
receipt, e := client.Execute(ctx, fresh, []byte("gateway fixture shared-fresh"))
if e != nil || receipt.State != "committed" {
t.Fatalf("planned create blocked by unrelated writers: %v %s", e, receipt.State)
}
rejected, e := client.Execute(ctx, clash, []byte("gateway fixture shared-clash"))
if e != nil || rejected.State != "rejected" {
t.Fatalf("pre-existing planned key not rejected: %v %s", e, rejected.State)
}
after, exists, e := backend.Observe(ctx, existingKey, "")
if e != nil || !exists || after.ID != existing.ID || after.SHA256 != g.Digest(existingBody) {
t.Fatalf("pre-existing object modified: %v %+v", e, after)
}
still, exists, e := backend.Observe(ctx, unrelatedKey, "")
if e != nil || !exists || still.ID != unrelated.ID {
t.Fatalf("unrelated object disturbed: %v %+v", e, still)
}
// Precise rollback removes only the version this gateway created.
cleanup(t, ctx, &g.Gateway{Pool: pool, Scope: scope, Backend: backend, MaxBytes: 1 << 20}, receipt)
for key, keep := range map[string]g.Version{existingKey: existing, unrelatedKey: unrelated} {
if e = backend.Delete(ctx, key, keep); e != nil {
t.Fatalf("external fixture cleanup %s: %v", key, e)
}
}
}
@@ -13,7 +13,7 @@ import (
func TestGatewayNewEpochPreservesOriginalSendIdentity(t *testing.T) {
ctx, pool, backend := fixture(t)
gateway := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20}
gateway := &g.Gateway{Pool: pool, Scope: backend.Bucket, Backend: backend, MaxBytes: 1 << 20}
first := g.Grant{Epoch: 1, Owner: g.Digest([]byte(t.Name())), Plan: g.Digest([]byte("takeover-plan"))}
sent, unsent := request("takeover-sent", first), request("takeover-unsent", first)
if e := gateway.InstallPlan(ctx, planFor(first, sent, unsent)); e != nil {
@@ -22,7 +22,7 @@ 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}
gateway := &g.Gateway{Pool: pool, Scope: real.Bucket, Backend: backend, MaxBytes: 1 << 20}
if e := gateway.InstallPlan(ctx, planFor(grant, request(t.Name(), grant))); e != nil {
t.Fatal(e)
}
@@ -62,7 +62,7 @@ func TestGatewayReadOnlyPersistenceBlocksSend(t *testing.T) {
t.Fatal(e)
}
defer other.Close()
gateway := &g.Gateway{Pool: other, Scope: backend.Bucket, Backend: backend, Isolation: verifiedFixture{}, MaxBytes: 1 << 20}
gateway := &g.Gateway{Pool: other, Scope: backend.Bucket, Backend: backend, MaxBytes: 1 << 20}
if e = gateway.Ready(ctx); !errors.Is(e, g.ErrPersistence) {
t.Fatal("read-only persistence accepted", e)
}
@@ -54,7 +54,7 @@ func OpenR10AdmissionFixture(t *testing.T) R10AdmissionFixture {
authority := &GatewayGovernanceAuthority{Pool: reader, ArchiveRoot: fx.archiveRoot, TargetID: "r10-admission", Provider: "minio", Bucket: fx.bucket}
endpoint := os.Getenv("KG_LIBRARY_TEST_S3_ENDPOINT")
scope := objectgateway.Digest([]byte("minio\x00" + endpoint + "\x00" + fx.bucket))
gateway := &objectgateway.Gateway{Pool: governancePool, Scope: scope, Backend: &objectgateway.S3Backend{Client: fx.objects.s3.client, Bucket: fx.bucket, MaxBytes: 1 << 20}, Authority: authority, Isolation: gatewayFixtureIsolation{}, MaxBytes: 1 << 20}
gateway := &objectgateway.Gateway{Pool: governancePool, Scope: scope, Backend: &objectgateway.S3Backend{Client: fx.objects.s3.client, Bucket: fx.bucket, MaxBytes: 1 << 20}, Authority: authority, MaxBytes: 1 << 20}
token := objectgateway.Digest([]byte("r10-admission-data"))
server := httptest.NewServer(objectgateway.HTTPServer{Gateway: gateway, DataToken: token, ControlToken: objectgateway.Digest([]byte("r10-admission-control"))})
t.Cleanup(server.Close)
@@ -128,7 +128,7 @@ func TestContractGatewayAuthorityFrozenPlanIsolated(t *testing.T) {
t.Fatal("privileged controller reader accepted")
}
backend := &gatewayCountBackend{Backend: &objectgateway.S3Backend{Client: fx.objects.s3.client, Bucket: fx.bucket, MaxBytes: 1 << 20}}
gateway := &objectgateway.Gateway{Pool: governancePool, Scope: fx.bucket, Backend: backend, Authority: authority, Isolation: gatewayFixtureIsolation{}, MaxBytes: 1 << 20}
gateway := &objectgateway.Gateway{Pool: governancePool, Scope: fx.bucket, Backend: backend, Authority: authority, MaxBytes: 1 << 20}
data := objectgateway.Digest([]byte("authority-data"))
server := httptest.NewServer(objectgateway.HTTPServer{Gateway: gateway, DataToken: data, ControlToken: objectgateway.Digest([]byte("authority-control"))})
defer server.Close()
@@ -9,10 +9,6 @@ import (
"knowledge-governance/internal/objectgateway"
)
type gatewayFixtureIsolation struct{}
func (gatewayFixtureIsolation) Verify(context.Context) error { return nil }
// Test controller issues the grant only after the saga has frozen its actual
// plan. Production controller delivery is a separate integration prerequisite.
type gatewayFixtureController struct {
@@ -35,7 +31,7 @@ func TestContractGatewayActualSagaRollbackIsolated(t *testing.T) {
before := universeDump(t, fx.pool)
snapshot := sealSourceSnapshot(t, fx, nil)
target := fx.target(t)
gateway := &objectgateway.Gateway{Pool: fx.pool, Scope: fx.bucket, Backend: &objectgateway.S3Backend{Client: fx.objects.s3.client, Bucket: fx.bucket, MaxBytes: 1 << 20}, Isolation: gatewayFixtureIsolation{}, MaxBytes: 1 << 20}
gateway := &objectgateway.Gateway{Pool: fx.pool, Scope: fx.bucket, Backend: &objectgateway.S3Backend{Client: fx.objects.s3.client, Bucket: fx.bucket, MaxBytes: 1 << 20}, MaxBytes: 1 << 20}
token := objectgateway.Digest([]byte("data-fixture"))
server := httptest.NewServer(objectgateway.HTTPServer{Gateway: gateway, DataToken: token, ControlToken: objectgateway.Digest([]byte("control-fixture"))})
defer server.Close()
+5
View File
@@ -45,6 +45,11 @@
"test": "TestGatewayAcquisitionPersistsAttemptDelivery",
"reason": "仅显式一次性 loopback 治理PG和随机MinIO测试桶运行;验证持久grant交付、跨epoch发送身份及unknown阻断,本机隔离证据单列,普通CI跳过不算通过"
},
{
"package": "knowledge-governance/internal/objectgateway",
"test": "TestGatewaySharedBucketCoexistenceWithOtherWriters",
"reason": "仅显式一次性loopback kg_library_test_gateway_*治理库、随机MinIO桶及本轮构建网关二进制执行:无整桶证明启动、无关key共存写入不阻塞、同key既有对象拒绝且不改字节、精确回滚;普通CI不带存储凭据,本机隔离实跑另存raw,跳过不算通过"
},
{
"package": "knowledge-governance/internal/objectgateway",
"test": "TestGatewayAcquisitionCannotHandOverUnknownSend",