A container registry that uses the AT Protocol for manifest storage and S3 for blob storage.
dedupe hold health checks
+132
-27
+1
-1
deploy/docker-compose.prod.yml
+1
-1
deploy/docker-compose.prod.yml
+2
-2
pkg/appview/db/hold_store.go
+2
-2
pkg/appview/db/hold_store.go
···
8
8
9
9
// HoldCaptainRecord represents a cached captain record from a hold's PDS
10
10
type HoldCaptainRecord struct {
11
-
HoldDID string `json:"-"` // Set manually, not from JSON
11
+
HoldDID string `json:"-"` // Set manually, not from JSON
12
12
OwnerDID string `json:"owner"`
13
13
Public bool `json:"public"`
14
14
AllowAllCrew bool `json:"allowAllCrew"`
15
15
DeployedAt string `json:"deployedAt"`
16
16
Region string `json:"region"`
17
17
Provider string `json:"provider"`
18
-
UpdatedAt time.Time `json:"-"` // Set manually, not from JSON
18
+
UpdatedAt time.Time `json:"-"` // Set manually, not from JSON
19
19
}
20
20
21
21
// GetCaptainRecord retrieves a captain record from the cache
+53
pkg/appview/holdhealth/checker_test.go
+53
pkg/appview/holdhealth/checker_test.go
···
268
268
t.Errorf("Expected startupDelay=%v, got %v", startupDelay, workerWithDelay.startupDelay)
269
269
}
270
270
}
271
+
272
+
func TestNormalizeHoldEndpoint(t *testing.T) {
273
+
tests := []struct {
274
+
name string
275
+
input string
276
+
expected string
277
+
}{
278
+
{
279
+
name: "HTTP URL",
280
+
input: "http://hold01.atcr.io",
281
+
expected: "did:web:hold01.atcr.io",
282
+
},
283
+
{
284
+
name: "HTTPS URL",
285
+
input: "https://hold01.atcr.io",
286
+
expected: "did:web:hold01.atcr.io",
287
+
},
288
+
{
289
+
name: "HTTP URL with port",
290
+
input: "http://172.28.0.3:8080",
291
+
expected: "did:web:172.28.0.3:8080",
292
+
},
293
+
{
294
+
name: "HTTP URL with trailing slash",
295
+
input: "http://hold01.atcr.io/",
296
+
expected: "did:web:hold01.atcr.io",
297
+
},
298
+
{
299
+
name: "HTTP URL with path",
300
+
input: "http://hold01.atcr.io/some/path",
301
+
expected: "did:web:hold01.atcr.io",
302
+
},
303
+
{
304
+
name: "Already a DID",
305
+
input: "did:web:hold01.atcr.io",
306
+
expected: "did:web:hold01.atcr.io",
307
+
},
308
+
{
309
+
name: "DID with port",
310
+
input: "did:web:172.28.0.3:8080",
311
+
expected: "did:web:172.28.0.3:8080",
312
+
},
313
+
}
314
+
315
+
for _, tt := range tests {
316
+
t.Run(tt.name, func(t *testing.T) {
317
+
result := normalizeHoldEndpoint(tt.input)
318
+
if result != tt.expected {
319
+
t.Errorf("normalizeHoldEndpoint(%q) = %q, want %q", tt.input, result, tt.expected)
320
+
}
321
+
})
322
+
}
323
+
}
+54
-2
pkg/appview/holdhealth/worker.go
+54
-2
pkg/appview/holdhealth/worker.go
···
5
5
"database/sql"
6
6
"fmt"
7
7
"log"
8
+
"strings"
8
9
"sync"
9
10
"time"
10
11
)
···
115
116
return
116
117
}
117
118
118
-
log.Printf("Hold health worker: Checking %d unique hold endpoints", len(endpoints))
119
+
log.Printf("Hold health worker: Fetched %d hold endpoint entries from database", len(endpoints))
120
+
121
+
// Deduplicate endpoints by normalizing to canonical DID format
122
+
// This handles cases where the same hold is stored with different representations:
123
+
// - http://172.28.0.3:8080 (internal IP)
124
+
// - http://hold01.atcr.io (external hostname)
125
+
// - did:web:hold01.atcr.io (DID format)
126
+
// All normalize to the same DID: did:web:hold01.atcr.io (or did:web:172.28.0.3:8080)
127
+
seen := make(map[string]bool)
128
+
uniqueEndpoints := make([]string, 0, len(endpoints))
129
+
130
+
for _, endpoint := range endpoints {
131
+
// Normalize to canonical DID format
132
+
normalizedDID := normalizeHoldEndpoint(endpoint)
133
+
134
+
// Skip if we've already seen this normalized DID
135
+
if seen[normalizedDID] {
136
+
continue
137
+
}
138
+
139
+
seen[normalizedDID] = true
140
+
// Use the normalized DID for health checks
141
+
uniqueEndpoints = append(uniqueEndpoints, normalizedDID)
142
+
}
143
+
144
+
log.Printf("Hold health worker: Checking %d unique hold endpoints (deduplicated from %d)", len(uniqueEndpoints), len(endpoints))
119
145
120
146
// Check health concurrently with rate limiting
121
147
// Use a semaphore to limit concurrent requests (max 10 at a time)
···
126
152
unreachable := 0
127
153
var statsMu sync.Mutex
128
154
129
-
for _, endpoint := range endpoints {
155
+
for _, endpoint := range uniqueEndpoints {
130
156
wg.Add(1)
131
157
132
158
go func(ep string) {
···
193
219
194
220
return endpoints, nil
195
221
}
222
+
223
+
// normalizeHoldEndpoint converts a hold endpoint (URL or DID) to canonical DID format
224
+
// This ensures that different representations of the same hold are deduplicated:
225
+
// - http://172.28.0.3:8080 → did:web:172.28.0.3:8080
226
+
// - http://hold01.atcr.io → did:web:hold01.atcr.io
227
+
// - https://hold01.atcr.io → did:web:hold01.atcr.io
228
+
// - did:web:hold01.atcr.io → did:web:hold01.atcr.io (passthrough)
229
+
func normalizeHoldEndpoint(endpoint string) string {
230
+
// Strip protocol and trailing slashes
231
+
normalized := endpoint
232
+
normalized = strings.TrimPrefix(normalized, "http://")
233
+
normalized = strings.TrimPrefix(normalized, "https://")
234
+
normalized = strings.TrimSuffix(normalized, "/")
235
+
236
+
// If already a DID, return as-is
237
+
if strings.HasPrefix(endpoint, "did:") {
238
+
return endpoint
239
+
}
240
+
241
+
// Extract hostname (remove path if present)
242
+
parts := strings.Split(normalized, "/")
243
+
hostname := parts[0]
244
+
245
+
// Convert to did:web
246
+
return "did:web:" + hostname
247
+
}
+22
-22
pkg/hold/pds/server.go
+22
-22
pkg/hold/pds/server.go
···
128
128
129
129
if !captainExists {
130
130
131
-
// Initialize repo if it doesn't exist yet
132
-
// Check if repo exists by trying to get the head
133
-
head, err := p.carstore.GetUserRepoHead(ctx, p.uid)
134
-
if err != nil || !head.Defined() {
135
-
// Repo doesn't exist, initialize it
136
-
// InitNewActor creates an empty repo with initial commit
137
-
err = p.repomgr.InitNewActor(ctx, p.uid, "", p.did, "", "", "")
138
-
if err != nil {
139
-
return fmt.Errorf("failed to initialize repo: %w", err)
131
+
// Initialize repo if it doesn't exist yet
132
+
// Check if repo exists by trying to get the head
133
+
head, err := p.carstore.GetUserRepoHead(ctx, p.uid)
134
+
if err != nil || !head.Defined() {
135
+
// Repo doesn't exist, initialize it
136
+
// InitNewActor creates an empty repo with initial commit
137
+
err = p.repomgr.InitNewActor(ctx, p.uid, "", p.did, "", "", "")
138
+
if err != nil {
139
+
return fmt.Errorf("failed to initialize repo: %w", err)
140
+
}
141
+
fmt.Printf("✅ Initialized empty repo\n")
140
142
}
141
-
fmt.Printf("✅ Initialized empty repo\n")
142
-
}
143
143
144
-
// Create captain record (hold ownership and settings)
145
-
_, err = p.CreateCaptainRecord(ctx, ownerDID, public, allowAllCrew)
146
-
if err != nil {
147
-
return fmt.Errorf("failed to create captain record: %w", err)
148
-
}
144
+
// Create captain record (hold ownership and settings)
145
+
_, err = p.CreateCaptainRecord(ctx, ownerDID, public, allowAllCrew)
146
+
if err != nil {
147
+
return fmt.Errorf("failed to create captain record: %w", err)
148
+
}
149
149
150
-
fmt.Printf("✅ Created captain record (public=%v, allowAllCrew=%v)\n", public, allowAllCrew)
150
+
fmt.Printf("✅ Created captain record (public=%v, allowAllCrew=%v)\n", public, allowAllCrew)
151
151
152
-
// Add hold owner as first crew member with admin role
153
-
_, err = p.AddCrewMember(ctx, ownerDID, "admin", []string{"blob:read", "blob:write", "crew:admin"})
154
-
if err != nil {
155
-
return fmt.Errorf("failed to add owner as crew member: %w", err)
156
-
}
152
+
// Add hold owner as first crew member with admin role
153
+
_, err = p.AddCrewMember(ctx, ownerDID, "admin", []string{"blob:read", "blob:write", "crew:admin"})
154
+
if err != nil {
155
+
return fmt.Errorf("failed to add owner as crew member: %w", err)
156
+
}
157
157
158
158
fmt.Printf("✅ Added %s as hold admin\n", ownerDID)
159
159
}