fix: 保持后端 assignment 可重试

internal/assignmentqueue/jetstream.go

@@ -167,7 +167,7 @@ func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectB
 		AckPolicy:     jetstream.AckExplicitPolicy,
 		AckWait:       5 * time.Minute,
 		MaxAckPending: capacity,
-		MaxDeliver:    20,
+		MaxDeliver:    1000,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err)

internal/assignmentqueue/jetstream_test.go

@@ -205,7 +205,7 @@ func TestOpenConsumerUsesIndependentDurablePerBackend(t *testing.T) {
 	if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.BackendPod, 4); err != nil {
 		t.Fatal(err)
 	}
-	if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 {
+	if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 {
 		t.Fatalf("config = %#v", manager.config)
 	}
 }

internal/opensandboxbackend/backend.go

@@ -3,9 +3,13 @@ package opensandboxbackend
 
 import (
 	"context"
+	"encoding/base64"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
+	"sort"
+	"strings"
 	"time"
 
 	opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go"
@@ -17,6 +21,8 @@ import (
 
 const assignmentMetadata = "ci.ddupan.top/assignment-id"
 const terminalMetadata = "ci.ddupan.top/terminal"
+const annotationsMetadataPrefix = "ci.ddupan.top/annotations-"
+const metadataValueLimit = 63
 
 type Lifecycle interface {
 	ListSandboxes(context.Context, opensandbox.ListOptions) (*opensandbox.ListSandboxesResponse, error)
@@ -123,7 +129,11 @@ func (b Backend) RecoverAssignments(ctx context.Context, trustDomain string) ([]
 		if sandbox.Metadata[terminalMetadata] == "true" {
 			continue
 		}
-		assignment, err := taskassignment.FromMetadata(sandbox.Metadata, sandbox.Metadata, trustDomain)
+		annotations, err := decodeAnnotations(sandbox.Metadata)
+		if err != nil {
+			return nil, fmt.Errorf("decode sandbox %s annotations: %w", sandbox.ID, err)
+		}
+		assignment, err := taskassignment.FromMetadata(sandbox.Metadata, annotations, trustDomain)
 		if err != nil {
 			return nil, fmt.Errorf("recover sandbox %s: %w", sandbox.ID, err)
 		}
@@ -169,8 +179,12 @@ func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignmen
 	for key, value := range launch.Environment {
 		environment[key] = value
 	}
-	sandboxMetadata := clone(launch.Metadata.Annotations)
+	sandboxMetadata := clone(launch.Metadata.Labels)
-	for key, value := range launch.Metadata.Labels {
+	annotations, err := encodeAnnotations(launch.Metadata.Annotations)
+	if err != nil {
+		return nil, fmt.Errorf("encode sandbox annotations: %w", err)
+	}
+	for key, value := range annotations {
 		sandboxMetadata[key] = value
 	}
 	request := opensandbox.CreateSandboxRequest{
@@ -194,7 +208,11 @@ func (b Backend) BindIdentity(ctx context.Context, executor *taskworker.Executor
 	if err != nil {
 		return fmt.Errorf("verify sandbox identity metadata: %w", err)
 	}
-	if sandbox.Metadata["ci.ddupan.top/spiffe-id"] != identity.SPIFFEID {
+	annotations, err := decodeAnnotations(sandbox.Metadata)
+	if err != nil {
+		return fmt.Errorf("decode sandbox identity metadata: %w", err)
+	}
+	if annotations["ci.ddupan.top/spiffe-id"] != identity.SPIFFEID {
 		return fmt.Errorf("sandbox %s has inconsistent SPIFFE identity metadata", executor.Name)
 	}
 	return nil
@@ -243,3 +261,50 @@ func clone(source map[string]string) map[string]string {
 	}
 	return result
 }
+
+// OpenSandbox metadata follows Kubernetes label-value constraints, unlike Pod
+// annotations. Store the annotation map as deterministic URL-safe base64
+// chunks so repository paths and SPIFFE IDs remain lossless and recoverable.
+func encodeAnnotations(annotations map[string]string) (map[string]string, error) {
+	data, err := json.Marshal(annotations)
+	if err != nil {
+		return nil, err
+	}
+	encoded := base64.RawURLEncoding.EncodeToString(data)
+	result := make(map[string]string, (len(encoded)+metadataValueLimit-1)/metadataValueLimit)
+	for index := 0; len(encoded) > 0; index++ {
+		length := min(metadataValueLimit, len(encoded))
+		result[fmt.Sprintf("%s%03d", annotationsMetadataPrefix, index)] = encoded[:length]
+		encoded = encoded[length:]
+	}
+	return result, nil
+}
+
+func decodeAnnotations(metadata map[string]string) (map[string]string, error) {
+	keys := make([]string, 0)
+	for key := range metadata {
+		if strings.HasPrefix(key, annotationsMetadataPrefix) {
+			keys = append(keys, key)
+		}
+	}
+	if len(keys) == 0 {
+		return nil, errors.New("sandbox annotation metadata is missing")
+	}
+	sort.Strings(keys)
+	var encoded strings.Builder
+	for index, key := range keys {
+		if key != fmt.Sprintf("%s%03d", annotationsMetadataPrefix, index) {
+			return nil, errors.New("sandbox annotation metadata chunks are incomplete")
+		}
+		encoded.WriteString(metadata[key])
+	}
+	data, err := base64.RawURLEncoding.DecodeString(encoded.String())
+	if err != nil {
+		return nil, err
+	}
+	var annotations map[string]string
+	if err := json.Unmarshal(data, &annotations); err != nil {
+		return nil, err
+	}
+	return annotations, nil
+}

internal/opensandboxbackend/backend_test.go

@@ -31,7 +31,8 @@ func (f *fakeLifecycle) GetSandbox(_ context.Context, id string) (*opensandbox.S
 			return &item, nil
 		}
 	}
-	return &opensandbox.SandboxInfo{ID: id, Metadata: map[string]string{"ci.ddupan.top/spiffe-id": assignment().Identity.SPIFFEID}}, nil
+	metadata, _ := encodeAnnotations(map[string]string{"ci.ddupan.top/spiffe-id": assignment().Identity.SPIFFEID})
+	return &opensandbox.SandboxInfo{ID: id, Metadata: metadata}, nil
 }
 func (f *fakeLifecycle) PatchSandboxMetadata(_ context.Context, id string, patch opensandbox.MetadataPatch) (*opensandbox.SandboxInfo, error) {
 	for index := range f.items {
@@ -89,12 +90,38 @@ func TestCreateUsesPoolAndPersistsRecoveryMetadata(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if lifecycle.created.Extensions["poolRef"] != "ci-vm" || lifecycle.created.Metadata[assignmentMetadata] != assignment().ID || lifecycle.created.Env["CI_SPIFFE_ID"] != assignment().Identity.SPIFFEID || executor.Name != "sandbox-42" {
+	if lifecycle.created.Extensions["poolRef"] != "ci-vm" || lifecycle.created.Metadata[assignmentMetadata] != assignment().ID || lifecycle.created.Metadata["ci.ddupan.top/runner"] != "true" || lifecycle.created.Env["CI_SPIFFE_ID"] != assignment().Identity.SPIFFEID || executor.Name != "sandbox-42" {
 		t.Fatalf("request=%#v executor=%#v", lifecycle.created, executor)
 	}
 	if lifecycle.created.Env["CI_RUNNER_CAPABILITY"] != "capability" {
 		t.Fatalf("environment = %#v", lifecycle.created.Env)
 	}
+	annotations, err := decodeAnnotations(lifecycle.created.Metadata)
+	if err != nil || annotations["ci.ddupan.top/repository"] != "owner/repo" || annotations["ci.ddupan.top/spiffe-id"] != assignment().Identity.SPIFFEID {
+		t.Fatalf("annotations=%#v err=%v", annotations, err)
+	}
+	for _, value := range lifecycle.created.Metadata {
+		if len(value) > metadataValueLimit {
+			t.Fatalf("metadata value exceeds %d characters: %q", metadataValueLimit, value)
+		}
+	}
+}
+
+func TestAnnotationMetadataRoundTripPreservesSlashValues(t *testing.T) {
+	want := taskworker.BackendMetadata(assignment()).Annotations
+	encoded, err := encodeAnnotations(want)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, err := decodeAnnotations(encoded)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for key, value := range want {
+		if got[key] != value {
+			t.Fatalf("%s=%q, want %q", key, got[key], value)
+		}
+	}
 }
 
 func TestBindIdentityVerifiesPersistedMetadata(t *testing.T) {

internal/taskworker/worker.go

@@ -178,6 +178,7 @@ func (w Worker) launchSpec(assignment taskassignment.Assignment) (LaunchSpec, er
 func BackendMetadata(assignment taskassignment.Assignment) Metadata {
 	return Metadata{
 		Labels: map[string]string{
+			"ci.ddupan.top/runner":        "true",
 			"ci.ddupan.top/assignment-id": assignment.ID,
 			"ci.ddupan.top/task-id":       strconv.FormatInt(assignment.Task.GetId(), 10),
 			"ci.ddupan.top/backend":       string(assignment.Backend),