Implement broker feature enhancements

.dockerignore

@@ -0,0 +1,16 @@
+.git/
+.github/
+.dockerignore
+Dockerfile*
+vendor/
+bin/
+coverage/
+*.out
+*.test
+*.prof
+*.log*
+.env*
+.DS_Store
+.idea/
+.vscode/
+specs/

.github/workflows/ci.yml

@@ -13,10 +13,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
 
       - name: Set up Go
-        uses: actions/setup-go@v5
+        uses: actions/setup-go@v6
         with:
           go-version: '1.24'
           cache: true
@@ -25,7 +25,7 @@ jobs:
         run: go mod verify
 
       - name: Lint
-        uses: golangci/golangci-lint-action@v6
+        uses: golangci/golangci-lint-action@v9
         with:
           version: latest
 

.github/workflows/integration.yml

@@ -16,10 +16,10 @@ jobs:
     timeout-minutes: 20
     steps:
       - name: Checkout code
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
 
       - name: Set up Go
-        uses: actions/setup-go@v5
+        uses: actions/setup-go@v6
         with:
           go-version: "1.24"
           cache: true
@@ -28,11 +28,38 @@ jobs:
         run: make integration-up
 
       - name: Run Docker integration tests
-        run: go test -tags=integration -count=1 -v ./integration
+        shell: bash
+        run: |
+          set -o pipefail
+          go test -tags=integration -count=1 -v ./integration 2>&1 | tee integration-test.log
+
+      - name: Reject sensitive observability output
+        if: always()
+        shell: bash
+        run: |
+          sed -E 's/((authorization|password|secret|token|api[-_]?key)[=:])[[:graph:]]+/\1[REDACTED]/Ig' integration-test.log > sanitized-integration-test.log
+          if grep -Eiq '(authorization|password|secret|token|api[-_]?key)[=:][^[:space:]]+' integration-test.log; then
+            echo 'Sensitive-looking value found in integration output'
+            exit 1
+          fi
 
       - name: Show container logs on failure
         if: failure()
-        run: docker compose -f docker-compose.integration.yml logs --no-color
+        shell: bash
+        run: |
+          docker compose -f docker-compose.integration.yml logs --no-color 2>&1 \
+            | sed -E 's/((authorization|password|secret|token|api[-_]?key)[=:])[[:graph:]]+/\1[REDACTED]/Ig' \
+            | tee sanitized-container.log
+
+      - name: Upload sanitized failure logs
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: sanitized-integration-logs
+          path: |
+            sanitized-integration-test.log
+            sanitized-container.log
+          if-no-files-found: ignore
 
       - name: Clean up containers
         if: always()

.github/workflows/pull-request.yml

@@ -18,10 +18,10 @@ jobs:
     timeout-minutes: 10
     steps:
       - name: Checkout code
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
 
       - name: Set up Go
-        uses: actions/setup-go@v5
+        uses: actions/setup-go@v6
         with:
           go-version: "1.24"
           cache: true
@@ -41,8 +41,17 @@ jobs:
       - name: Run go vet
         run: go vet ./...
 
+      - name: Build observability example
+        run: go build ./examples/observability
+
+      - name: Smoke test observability switches
+        run: |
+          go run ./examples/observability
+          go run ./examples/observability -observe -health -diagnostics
+          go run ./examples/observability -observe -metrics -correlation -events
+
       - name: Run golangci-lint
-        uses: golangci/golangci-lint-action@v6
+        uses: golangci/golangci-lint-action@v9
         with:
           version: latest
 
@@ -52,10 +61,10 @@ jobs:
     timeout-minutes: 15
     steps:
       - name: Checkout code
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
 
       - name: Set up Go
-        uses: actions/setup-go@v5
+        uses: actions/setup-go@v6
         with:
           go-version: "1.24"
           cache: true
@@ -70,5 +79,10 @@ jobs:
           awk -v total="$total" 'BEGIN { if (total < 75) { printf "coverage %.1f%% is below 75%%\n", total; exit 1 } }'
           printf 'total coverage: %s%%\n' "$total"
 
+      - name: Enforce observability performance budget
+        env:
+          BROKER_ENFORCE_PERFORMANCE: "1"
+        run: go test -count=1 -run '^TestObservabilityStandardOverheadBudget$' .
+
       - name: Run race detector
         run: go test -race -count=1 ./...

.golangci.yml

@@ -1,43 +1,46 @@
+version: "2"
+
 run:
   timeout: 5m
   issues-exit-code: 1
   tests: true
+  allow-parallel-runners: true
 
 linters:
+  default: none
   enable:
     - errcheck
-    - gosimple
     - staticcheck
     - unused
     - govet
     - ineffassign
-    - typecheck
-
-issues:
-  # Exclude some common false positives or errors that are acceptable in specific contexts
-  exclude-rules:
-    # 1. Allow ignoring errors in test files (test code is usually less strict)
-    - path: '_test\.go'
-      linters:
-        - errcheck
-        - staticcheck
-      text: "(SA1012|Error return value of .*(Init|Connect). is not checked)"
+  exclusions:
+    generated: lax
+    # Exclude some common false positives or errors that are acceptable in specific contexts
+    rules:
+      # 1. Allow ignoring errors in test files (test code is usually less strict)
+      - path: '_test\.go'
+        linters:
+          - errcheck
+          - staticcheck
+        text: "(SA1012|Error return value of .*(Init|Connect). is not checked)"
 
-    # 2. Ignore errcheck for specific methods where missing error check is common/acceptable
-    - linters:
-        - errcheck
-      text: 'Error return value of .((os\.)?std(out|err)\..*|.*Close|.*Shutdown|.*Unsubscribe|.*Ack|.*Nack). is not checked'
+      # 2. Ignore errcheck for specific methods where missing error check is common/acceptable
+      - linters:
+          - errcheck
+        text: 'Error return value of .((os\.)?std(out|err)\..*|.*Close|.*Shutdown|.*Unsubscribe|.*Ack|.*Nack). is not checked'
 
-    # 3. Ignore deprecation warnings for now (planned for future upgrades)
-    - linters:
-        - staticcheck
-      text: "SA1019"
+      # 3. Ignore deprecation warnings for now (planned for future upgrades)
+      - linters:
+          - staticcheck
+        text: "SA1019"
 
-    # 4. Ignore warnings about using built-in string type as context keys (to be fixed later)
-    - linters:
-        - staticcheck
-      text: "SA1029"
+      # 4. Ignore warnings about using built-in string type as context keys (to be fixed later)
+      - linters:
+          - staticcheck
+        text: "SA1029"
 
+issues:
   # Set max issues to 0 to show all findings and avoid suppression during initial setup
   max-issues-per-linter: 0
   max-same-issues: 0

.specify/feature.json

@@ -0,0 +1,3 @@
+{
+  "feature_directory": "specs/006-observability-diagnostics"
+}

ADAPTER_EXTENSIONS.md

@@ -175,3 +175,25 @@
 | `redis.WithMaxLen(int64)` | 设置 Stream 的最大长度 (`MAXLEN`) |
 
 **注**：Redis 适配器基于 **Redis Streams** 实现。订阅时必须通过 `broker.Queue(groupName)` 指定 Consumer Group 名称以保证消息可靠消费（PEL 支持）。
+
+---
+
+## 10. 可观测能力矩阵
+
+所有内置适配器都通过可选 `broker.Observable` 接口提供本地生命周期状态、不可变诊断快照、发布/Handler/Ack/Nack 结果、固定维度指标和 W3C 上下文传递。该能力不会加入 `broker.Broker`，第三方实现无需修改。
+
+| 适配器 | 被动健康 | 主动探测 | 诊断/指标 | 上下文传递 | 验证方式 |
+| :--- | :--- | :--- | :--- | :--- | :--- |
+| Kafka | 支持 | 可取消 TCP dial | 支持 | Header | Docker 集成 |
+| RabbitMQ | 支持 | 连接状态检查 | 支持 | AMQP Header | Docker 集成 |
+| NATS | 支持 | `FlushWithContext` | 支持 | Header | Docker 集成 |
+| Redis Streams | 支持 | `PING` | 支持 | Stream 字段 | Docker 集成 |
+| RocketMQ | 支持 | 暂不支持 | 支持 | Message Property | Mock；发布前需真实服务复验 |
+| AWS SQS | 支持 | 暂不支持 | 支持 | Message Attribute | Mock；发布前需真实 AWS 队列复验 |
+| GCP Pub/Sub | 支持 | 暂不支持 | 支持 | Message Attribute | Mock；发布前需真实 GCP 项目复验 |
+
+被动 `Health()` 只读取本地同步状态，不发起网络请求。`Probe(ctx)` 仅在表中明确支持时执行，并遵守调用方取消和超时；其他适配器返回可用 `errors.Is` 判断的 `broker.ErrUnsupported`。所有能力默认关闭，启用方式和故障场景见 `examples/observability/`。
+
+### 云适配器发布前验证
+
+`go test -count=1 ./integration -run CloudAdapterObservabilityContracts` 会在无云凭证时验证 RocketMQ、SQS 与 Pub/Sub 的被动能力、固定信号集合、上下文复制和“不支持主动探测”契约。发布版本还应分别在隔离的真实 RocketMQ NameServer、AWS 测试队列和 GCP 测试项目中运行各适配器包测试，并人工确认：连接状态可恢复、消息正文和凭证未出现在输出中、Trace Context 能随消息往返、关闭后无后台消费者残留。真实服务凭证仅通过环境或工作负载身份注入，不写入命令、日志或仓库。