fix(docker,importer): scope init cleanup to project, Appwrite paging limits (D-0687, D-0784)

.github/wiki/Come-Home-Appwrite.md

@@ -66,7 +66,13 @@ column is NULL and the report lists the attribute and the number of documents un
 `exemptions` (id ending `#missing:<attribute>`); check those before you trust the data.
 Buckets are created in nSelf only for the objects they hold, so an empty bucket is listed in
 the report instead. A file name containing `/` or `\` has it replaced by `_`, and the file is
-listed in the report. Each downloaded file must match Appwrite's `sizeOriginal`.
+listed in the report. Each downloaded file must match Appwrite's `sizeOriginal`
+and its MD5 `signature` when the server supplies one.
+
+List pages start at 100 items and shrink on responses exceeding 64 MiB, down to
+one item. A single item larger than this limit stops the import. The plan uses a
+one-user count query; when its total reaches Appwrite's 5000-user cap, it walks
+the user list to count users, so planning large projects requires more requests.
 
 ## Passwords
 

internal/docker/cleanup.go

@@ -12,22 +12,23 @@ import (
 )
 
 // initContainerListArgs is the `docker ps` argument list that selects init
-// containers safe to remove: labelled nself.type=init-container, exited, with
+// containers safe to remove in the given project: labelled nself.type=init-container, exited, with
 // exit code 0. Running ones (a start in progress, or a service that uses the
 // label) and failed ones (kept for `docker logs`) are left alone.
-func initContainerListArgs() []string {
+func initContainerListArgs(projectName string) []string {
 	return []string{"ps", "-aq",
 		"--filter", "label=nself.type=init-container",
+		"--filter", "label=com.docker.compose.project=" + projectName,
 		"--filter", "status=exited",
 		"--filter", "exited=0",
 	}
 }
 
 // cleanupInitContainers removes finished init containers (minio-init,
-// meilisearch-init, seaweedfs-init, ...): the ones that exited with code 0.
+// meilisearch-init, seaweedfs-init, ...) in the given project that exited with code 0.
 // Running and failed init containers are kept.
-func cleanupInitContainers(ctx context.Context) error {
-	out, err := exec.CommandContext(ctx, "docker", initContainerListArgs()...).Output()
+func cleanupInitContainers(ctx context.Context, projectName string) error {
+	out, err := exec.CommandContext(ctx, "docker", initContainerListArgs(projectName)...).Output()
 	if err != nil {
 		return fmt.Errorf("finding init containers: %w", err)
 	}
@@ -114,7 +115,7 @@ func RunPreStartCleanup(ctx context.Context, projectName string) error {
 func RunPostStartCleanup(ctx context.Context, projectName string) error {
 	// Run init container cleanup 3 times with delays to handle race conditions.
 	for i := 0; i < 3; i++ {
-		if err := cleanupInitContainers(ctx); err != nil {
+		if err := cleanupInitContainers(ctx, projectName); err != nil {
 			// Non-fatal: continue with remaining passes.
 			slog.Debug("cleanup init containers (non-fatal)", "pass", i+1, "err", err)
 		}

internal/docker/cleanup_init_test.go

@@ -5,17 +5,74 @@ package docker
 //          (kept for `docker logs`) must never be selected.
 
 import (
+	"context"
+	"os"
+	"path/filepath"
 	"reflect"
+	"runtime"
+	"strings"
 	"testing"
 )
 
+func TestPostStartCleanupScopesInitContainersToProject(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("fake docker is a shell script")
+	}
+	dir := t.TempDir()
+	log := filepath.Join(dir, "calls")
+	t.Setenv("CLEANUP_CALLS", log)
+	t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	// Model two projects: an unscoped init query selects both; a scoped
+	// query selects only project-a. Other cleanup queries return no IDs.
+	script := `#!/bin/sh
+case " $* " in
+  *" label=nself.type=init-container "*)
+    printf 'init %s\n' "$*" >> "$CLEANUP_CALLS"
+    printf 'project-a-init\n'
+    case " $* " in
+      *" label=com.docker.compose.project=project-a "*) ;;
+      *) printf 'project-b-init\n' ;;
+    esac ;;
+  " rm -f "*) printf '%s\n' "$*" >> "$CLEANUP_CALLS" ;;
+esac
+`
+	if err := os.WriteFile(filepath.Join(dir, "docker"), []byte(script), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := RunPostStartCleanup(context.Background(), "project-a"); err != nil {
+		t.Fatal(err)
+	}
+	out, err := os.ReadFile(log)
+	if err != nil {
+		t.Fatal(err)
+	}
+	initQueries, removals := 0, 0
+	for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
+		if strings.HasPrefix(line, "init ") {
+			initQueries++
+			if !strings.Contains(line, "--filter label=com.docker.compose.project=project-a") {
+				t.Errorf("init query missing project filter: %s", line)
+			}
+		} else {
+			removals++
+			if line != "rm -f project-a-init" {
+				t.Errorf("removed another project's container: %s", line)
+			}
+		}
+	}
+	if initQueries != 3 || removals != 3 {
+		t.Errorf("got %d init queries and %d removals, want 3 each", initQueries, removals)
+	}
+}
+
 func TestInitContainerListArgsSelectExitedZeroOnly(t *testing.T) {
 	want := []string{"ps", "-aq",
 		"--filter", "label=nself.type=init-container",
+		"--filter", "label=com.docker.compose.project=project-a",
 		"--filter", "status=exited",
 		"--filter", "exited=0",
 	}
-	if got := initContainerListArgs(); !reflect.DeepEqual(got, want) {
+	if got := initContainerListArgs("project-a"); !reflect.DeepEqual(got, want) {
 		t.Fatalf("initContainerListArgs() = %v, want %v", got, want)
 	}
 }

internal/importer/appwrite/client.go

@@ -8,6 +8,7 @@ package appwrite
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -25,6 +26,8 @@ const (
 	maxPageBytes = 64 << 20
 )
 
+var errPageOverflow = errs.New("E517", "source schema lookup failed")
+
 // client is a minimal Appwrite REST client. The key is held only in the request header.
 type client struct {
 	base, project, key string
@@ -92,7 +95,14 @@ func (c *client) getJSON(ctx context.Context, path string, q url.Values, out any
 		return err
 	}
 	defer func() { _ = resp.Body.Close() }()
-	if json.NewDecoder(io.LimitReader(resp.Body, maxPageBytes)).Decode(out) != nil {
+	b, err := io.ReadAll(io.LimitReader(resp.Body, maxPageBytes+1))
+	if err != nil {
+		return errs.New("E517", "source schema lookup failed")
+	}
+	if len(b) > maxPageBytes {
+		return errPageOverflow
+	}
+	if json.Unmarshal(b, out) != nil {
 		return errs.New("E517", "source schema lookup failed")
 	}
 	return nil
@@ -108,13 +118,21 @@ func queryJSON(method string, values ...any) string {
 type page struct {
 	Total int64
 	Items []json.RawMessage
+	Limit int
 }
 
 // listPage fetches one page of the collection under key. A non-empty after is a $id cursor;
 // a positive offset pages by position instead (attribute documents carry no $id).
 func (c *client) listPage(ctx context.Context, path, key, after string, offset int) (page, error) {
+	return c.listPageLimit(ctx, path, key, after, offset, pageSize)
+}
+
+// listPageLimit retries oversized responses at the same position, down to one item.
+// Inputs: list location and requested limit. Outputs: items and the effective limit.
+// Constraints: only byte-cap overflow is retried; other source failures are returned.
+func (c *client) listPageLimit(ctx context.Context, path, key, after string, offset, limit int) (page, error) {
 	q := url.Values{}
-	q.Add("queries[]", queryJSON("limit", pageSize))
+	q.Add("queries[]", queryJSON("limit", limit))
 	if after != "" {
 		q.Add("queries[]", queryJSON("cursorAfter", after))
 	}
@@ -123,9 +141,12 @@ func (c *client) listPage(ctx context.Context, path, key, after string, offset i
 	}
 	var raw map[string]json.RawMessage
 	if err := c.getJSON(ctx, path, q, &raw); err != nil {
+		if errors.Is(err, errPageOverflow) && limit > 1 {
+			return c.listPageLimit(ctx, path, key, after, offset, limit/2)
+		}
 		return page{}, err
 	}
-	var p page
+	p := page{Limit: limit}
 	if t, ok := raw["total"]; ok && json.Unmarshal(t, &p.Total) != nil {
 		return page{}, errs.New("E517", "source schema lookup failed")
 	}
@@ -143,19 +164,20 @@ func (c *client) each(ctx context.Context, path, key string, fn func(json.RawMes
 
 // walk is each with a choice of cursor ($id) or offset paging.
 func (c *client) walk(ctx context.Context, path, key string, cursor bool, fn func(json.RawMessage) error) (int64, error) {
-	after, total, seen := "", int64(0), 0
+	after, total, seen, limit := "", int64(0), 0, pageSize
 	for first := true; ; first = false {
 		offset, prev := 0, after
 		if !cursor {
 			offset = seen
 		}
-		p, err := c.listPage(ctx, path, key, after, offset)
+		p, err := c.listPageLimit(ctx, path, key, after, offset, limit)
 		if err != nil {
 			return 0, err
 		}
 		if first {
 			total = p.Total
 		}
+		limit = p.Limit
 		for _, item := range p.Items {
 			if cursor {
 				var id struct {
@@ -178,7 +200,7 @@ func (c *client) walk(ctx context.Context, path, key string, cursor bool, fn fun
 		if !cursor && int64(seen) > total+pageSize {
 			return 0, errs.New("E517", "source schema lookup failed")
 		}
-		if len(p.Items) < pageSize {
+		if len(p.Items) < limit {
 			return total, nil
 		}
 	}

internal/importer/appwrite/count.go

@@ -0,0 +1,22 @@
+package appwrite
+
+import (
+	"context"
+	"encoding/json"
+)
+
+// userCount reads a cheap one-user query for Detect, then walks the list if its
+// total reaches Appwrite's count cap. Inputs: source client. Outputs: exact user
+// count. Constraints: capped totals cannot be trusted; no usage scope is required.
+func (c *client) userCount(ctx context.Context) (int64, error) {
+	p, err := c.listPageLimit(ctx, "/users", "users", "", 0, 1)
+	if err != nil || p.Total < totalCap {
+		return p.Total, err
+	}
+	var count int64
+	_, err = c.each(ctx, "/users", "users", func(json.RawMessage) error {
+		count++
+		return nil
+	})
+	return count, err
+}