-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathrun_experiment.py
More file actions
executable file
·4354 lines (3756 loc) · 182 KB
/
run_experiment.py
File metadata and controls
executable file
·4354 lines (3756 loc) · 182 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = ["pygal"]
# ///
"""
LFS Transfer Benchmark: compare git-lfs-transfer variants and storage backends.
Usage:
python3 run_experiment.py [--dataset small|r0] [--scenarios 0,1,...,11]
[--skip-build] [--skip-setup] [--keep-containers]
[--data-dir /workspace/desync/data/R0.git]
Scenarios:
0 plain git (no LFS)
1 baseline git-lfs-transfer (server-side, local files, scutiger or charmbracelet)
2 desync git-lfs-transfer (server-side, local chunks)
3 desync git-lfs-transfer (server-side, MinIO S3)
4 desync git-lfs-transfer (server-side, Garage S3)
5 git-lfs-desync client agent (client-side, MinIO S3)
6 git-lfs-desync client agent (client-side, Garage S3)
7 Stock Forgejo HTTP batch API (server-side, MinIO S3, upstream v14)
8 Forgejo HTTP batch API (server-side, MinIO S3, custom)
9 Forgejo HTTP batch API (server-side, desync Garage, custom)
10 Forgejo SSH pkt-line (server-side, desync Garage, custom)
11 Forgejo SSH pkt-line (server-side, MinIO S3, custom)
Cross-compatibility tests (after all scenarios):
3→client: clone scenario-3 data using client-side desync agent
4→client: clone scenario-4 data using client-side desync agent
5→server: clone scenario-5 data using server-side desync transfer
6→server: clone scenario-6 data using server-side desync transfer
9→client: clone scenario-9 data using client-side desync agent
10→http: clone scenario-10 SSH data using HTTP batch API
"""
import argparse
import hashlib
import hmac
import http.client
import json
import os
import random
import re
import shutil
import socket
import subprocess
import sys
import time
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import urllib.parse
import urllib.request
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
EXPERIMENT_DIR = Path(__file__).parent.resolve()
BINARIES_DIR = EXPERIMENT_DIR / "binaries"
# Source directories (git submodules).
DESYNC_SRC = EXPERIMENT_DIR / "desync"
SCUTIGER_SRC = EXPERIMENT_DIR / "scutiger"
CHARM_SRC = EXPERIMENT_DIR / "charm-git-lfs-transfer"
GIT_LFS_SRC = EXPERIMENT_DIR / "git-lfs"
FORGEJO_SRC = EXPERIMENT_DIR / "forgejo"
KEYS_DIR = EXPERIMENT_DIR / "keys"
SSH_KEY = KEYS_DIR / "lfs-test-key"
SSH_PUBKEY = KEYS_DIR / "lfs-test-key.pub"
GARAGE_TOML = EXPERIMENT_DIR / "garage.toml"
# Service coordinates (from host perspective, for management operations)
SSH_PORT = 2222
MINIO_USER = "minioadmin"
MINIO_PASS = "minioadmin"
# Shared S3 services (one MinIO + one Garage, wiped between scenarios)
SHARED_MINIO_CONTAINER = "lfs-exp-shared-minio"
SHARED_MINIO_ENDPOINT = "http://localhost:9000"
SHARED_MINIO_INTERNAL = "http://shared-minio:9000"
SHARED_GARAGE_CONTAINER = "lfs-exp-shared-garage"
SHARED_GARAGE_ENDPOINT = "http://localhost:3900"
SHARED_GARAGE_INTERNAL = "http://shared-garage:3900"
GIT_SSH_INTERNAL = "git-ssh" # docker service hostname
# Client-side desync config file path (inside client container).
# Written by setup_containers() with local-only settings (cache, concurrency).
# Uses the default config location so the agent loads it automatically.
# Server config (URLs, credentials, chunk-size) is provided separately via
# the SSH transfer negotiation protocol; MergeServerConfig() combines both.
CLIENT_DESYNC_CONFIG = "/home/dev/.config/desync/config.json"
# Client-side desync chunk cache path (inside client container).
# Must be on the ext4 /work bind mount, NOT on /tmp (overlayfs), because
# overlayfs copy-on-write is slow for many small file writes.
CLIENT_DESYNC_CACHE = "/work/.desync-cache"
# Docker container names (as set in docker-compose.yml)
GIT_SSH_CONTAINER = "lfs-exp-git-ssh"
CLIENT_CONTAINER = "lfs-exp-client"
S8_FORGEJO_CONTAINER = "lfs-exp-s8-forgejo"
S9_FORGEJO_CONTAINER = "lfs-exp-s9-forgejo"
FORGEJO_DB_CONTAINER = "lfs-exp-forgejo-db"
S10_FORGEJO_CONTAINER = "lfs-exp-s10-forgejo"
S7_FORGEJO_CONTAINER = "lfs-exp-s7-forgejo"
S8_FORGEJO_HTTP = "http://s8-forgejo:3000" # inside docker network
S9_FORGEJO_HTTP = "http://s9-forgejo:3000"
S10_FORGEJO_HTTP = "http://s10-forgejo:3000"
S7_FORGEJO_HTTP = "http://s7-forgejo:3000"
S11_FORGEJO_CONTAINER = "lfs-exp-s11-forgejo"
S12_FORGEJO_CONTAINER = "lfs-exp-s12-forgejo"
S13_GITEA_CONTAINER = "lfs-exp-s13-gitea"
S14_GITEA_CONTAINER = "lfs-exp-s14-gitea"
FORGEJO_USER = "testadmin"
FORGEJO_PASS = "admin12345678"
FORGEJO_EMAIL = "admin@test.local"
# Docker Compose project name (directory name); volumes are prefixed with this.
# Docker Compose derives the project name from the directory name (lowercased,
# keeping alphanumerics, hyphens, and underscores).
COMPOSE_PROJECT = re.sub(r"[^a-z0-9_-]", "", EXPERIMENT_DIR.name.lower())
# Path inside the client container where the host work dir is mounted
WORK_DIR_IN_CLIENT = "/work"
# SSH command for host→server connections (repo creation, config, etc.)
SSH_CMD = (
f"ssh -i {SSH_KEY} -o StrictHostKeyChecking=no"
f" -o UserKnownHostsFile=/dev/null -o BatchMode=yes"
)
# Set in main() before running scenarios; used by client path helpers.
WORK_DIR: Path = None
# Concurrency tuning. All settings can be overridden via environment variables.
# GIT_LFS_CONCURRENCY: how many LFS objects Git LFS transfers in parallel
# (each opens a separate SSH connection with its own git-lfs-transfer process).
GIT_LFS_CONCURRENCY = int(os.environ.get("GIT_LFS_CONCURRENCY", 8))
# DESYNC_CONCURRENCY: chunk workers inside each git-lfs-transfer / git-lfs-desync process.
DESYNC_CONCURRENCY = int(os.environ.get("DESYNC_CONCURRENCY", 10))
# Chunk size: min:avg:max in KB. Default is "64:256:1024".
CHUNK_SIZE = os.environ.get("CHUNK_SIZE", "64:256:1024")
# Maximum total bytes allowed in-flight across all concurrent transfer processes.
# Set to 0 to disable the limit. This caps memory usage when processing large
# datasets with multiple concurrent transfers.
MAX_IN_FLIGHT = int(os.environ.get("MAX_IN_FLIGHT", 250 * 1024**2)) # 250 MB
# Maximum concurrent storage operations (GetChunk/StoreChunk/GetIndex/StoreIndex)
# across all concurrent transfer processes. Set to 0 to disable.
MAX_STORAGE_OPS = int(os.environ.get("MAX_STORAGE_OPS", 20))
# Baseline (S1) git-lfs-transfer implementation: "scutiger" (Rust, default,
# used as reference by git-lfs test suite) or "charmbracelet" (Go).
S1_LFS_TRANSFER = os.environ.get("S1_LFS_TRANSFER", "scutiger")
# Verbose mode: set by --verbose flag, disables output capture so all
# subprocess output is visible for debugging.
VERBOSE = False
# CLI fragments for agent args (empty when disabled).
_MAX_IN_FLIGHT_ARG = f" --max-in-flight {MAX_IN_FLIGHT}" if MAX_IN_FLIGHT > 0 else ""
_MAX_STORAGE_OPS_ARG = f" --max-storage-ops {MAX_STORAGE_OPS}" if MAX_STORAGE_OPS > 0 else ""
# How often to reclaim page cache during push loops (every N refspec pushes).
# Uses cgroup v2 memory.reclaim to evict file-backed pages. Set to 0 to disable.
RECLAIM_INTERVAL = 0 # disabled
# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------
@dataclass
class ScenarioResult:
id: int
name: str
push_time_s: float = 0.0
clone_time_s: float = 0.0
storage_bytes: int = 0
raw_lfs_bytes: int = 0
# Network bytes [rx, tx] per container per operation.
push_net_bytes: list = field(default_factory=lambda: [-1, -1])
clone_net_bytes: list = field(default_factory=lambda: [-1, -1])
push_ssh_net_bytes: list = field(default_factory=lambda: [-1, -1])
clone_ssh_net_bytes: list = field(default_factory=lambda: [-1, -1])
push_s3_net_bytes: list = field(default_factory=lambda: [-1, -1])
clone_s3_net_bytes: list = field(default_factory=lambda: [-1, -1])
# S3 container measured (for display / JSON)
s3_container: str = ""
# LFS object counts: how many unique objects exist in .git/lfs/objects/
source_lfs_count: int = -1 # in source_repo (set per scenario)
clone_lfs_count: int = -1 # in clone after git lfs fetch --all
verified: bool = False
error: Optional[str] = None
notes: str = ""
# Resource usage per container: {container_name: ResourceSummary}
resources: dict = field(default_factory=dict)
@property
def storage_efficiency(self) -> float:
if self.raw_lfs_bytes == 0:
return 0.0
return self.storage_bytes / self.raw_lfs_bytes
@property
def storage_ratio(self) -> float:
"""Inverse of efficiency: raw bytes per stored byte. Higher = better."""
if self.storage_bytes <= 0:
return 0.0
return self.raw_lfs_bytes / self.storage_bytes
@dataclass
class CrossCompatResult:
name: str
clone_time_s: float = 0.0
verified: bool = False
error: Optional[str] = None
@dataclass
class ResourceSnapshot:
cpu_percent: float
mem_usage_bytes: int
mem_limit_bytes: int
mem_anon_bytes: int = 0 # RSS (process heap, stacks, mmap)
mem_cache_bytes: int = 0 # page cache (file-backed)
@dataclass
class ResourceSummary:
avg_cpu_percent: float = 0.0
max_cpu_percent: float = 0.0
avg_mem_bytes: int = 0
max_mem_bytes: int = 0
max_anon_bytes: int = 0
max_cache_bytes: int = 0
mem_limit_bytes: int = 0
sample_count: int = 0
# ---------------------------------------------------------------------------
# Logging helpers
# ---------------------------------------------------------------------------
RED = "\033[91m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
BLUE = "\033[94m"
CYAN = "\033[96m"
BOLD = "\033[1m"
RESET = "\033[0m"
def log(msg, color=RESET):
print(f"{color}{msg}{RESET}", flush=True)
def log_step(msg):
log(f"\n[*] {msg}", BOLD + CYAN)
def log_ok(msg):
log(f" [OK] {msg}", GREEN)
def log_warn(msg):
log(f" [WARN] {msg}", YELLOW)
def log_err(msg):
log(f" [ERR] {msg}", RED)
# ---------------------------------------------------------------------------
# Subprocess helpers
# ---------------------------------------------------------------------------
def run(cmd, cwd=None, env=None, check=True, capture=False, hide_output=False,
input=None, timeout=600):
"""Run a command, return CompletedProcess.
capture: capture stdout/stderr for the caller to read (result.stdout).
hide_output: suppress stdout/stderr (overridden by --verbose).
"""
do_capture = capture or (hide_output and not VERBOSE)
extra_env = None
if env:
extra_env = os.environ.copy()
extra_env.update(env)
result = subprocess.run(
cmd,
cwd=str(cwd) if cwd else None,
env=extra_env,
capture_output=do_capture,
text=True,
input=input,
timeout=timeout,
)
if check and result.returncode != 0:
stderr = result.stderr if do_capture else "(not captured)"
raise RuntimeError(
f"Command failed (exit {result.returncode}): {' '.join(str(c) for c in cmd)}\n{stderr}"
)
return result
def run_ssh(cmd_str, timeout=120, check=True):
"""Run a command on the git-ssh container as the git user."""
return run(
["ssh", "-i", str(SSH_KEY),
"-o", "StrictHostKeyChecking=no",
"-o", "UserKnownHostsFile=/dev/null",
"-o", "BatchMode=yes",
"-p", str(SSH_PORT),
f"git@localhost", cmd_str],
capture=True, check=check, timeout=timeout,
)
def compose(args, hide_output=True, env=None, check=True, timeout=120):
"""Run docker compose command in the experiment directory.
Output is hidden by default to keep the log clean (visible with --verbose).
"""
return run(
["docker", "compose"] + args,
cwd=EXPERIMENT_DIR,
hide_output=hide_output,
env=env,
check=check,
timeout=timeout,
)
def compose_exec(service, cmd_str, hide_output=True, check=True, timeout=120):
"""Run a command in a running docker compose service container."""
return compose(
["exec", "-T", service, "sh", "-c", cmd_str],
hide_output=hide_output,
check=check,
timeout=timeout,
)
# ---------------------------------------------------------------------------
# Client container helpers
# ---------------------------------------------------------------------------
def host_path_to_client(host_path: Path) -> str:
"""Convert a host path (inside WORK_DIR) to its path in the client container."""
rel = host_path.relative_to(WORK_DIR)
return f"{WORK_DIR_IN_CLIENT}/{rel}"
def client_exec(cmd: list, cwd: str = None, env: dict = None,
capture=False, check=True, timeout=3600) -> subprocess.CompletedProcess:
"""Run a command inside the client container via docker exec."""
dc = ["docker", "exec"]
if cwd:
dc += ["--workdir", cwd]
if env:
for k, v in env.items():
dc += ["-e", f"{k}={v}"]
dc += [CLIENT_CONTAINER] + cmd
return run(dc, capture=capture, hide_output=not capture, check=check, timeout=timeout)
def _batches(lst: list, size: int):
"""Yield successive non-overlapping slices of `lst` with at most `size` items."""
for i in range(0, len(lst), size):
yield lst[i:i + size]
def _pick_preferred_branch(branches: list) -> str:
"""Pick the best branch, preferring master/main, then alphabetical."""
for b in sorted(branches):
if "master" in b or "main" in b:
return b
return sorted(branches)[0]
def get_ordered_push_refspecs(source_repo: Path) -> list:
"""Return grouped push refspecs simulating a realistic CI workflow.
Returns a list of lists: each inner list is a group of refspecs to
push together in one ``git push`` invocation.
For tags: pushes the tag itself plus the preferred containing branch
(preferring master/main). For branches: pushes the branch update
unless it was already pushed to the same hash by a tag push.
"""
result = run(
["git", "for-each-ref", "refs/heads/*", "refs/tags/*",
"--format=%(creatordate:unix) %(objectname) %(refname)",
"--sort=creatordate"],
cwd=source_repo, capture=True,
)
refs = []
for line in result.stdout.strip().splitlines():
if not line:
continue
ts, sha, refname = line.split(None, 2)
refs.append((int(ts), sha, refname))
# Stable-sort: tags before branches at the same timestamp.
refs.sort(key=lambda r: (r[0], 0 if r[2].startswith("refs/tags/") else 1))
branch_pushed = {} # branch name -> last pushed sha
push_groups = []
for _ts, sha, refname in refs:
if refname.startswith("refs/tags/"):
tag_name = refname[len("refs/tags/"):]
group = [f"refs/tags/{tag_name}:refs/tags/{tag_name}"]
# Also update the preferred branch containing this commit.
r = run(
["git", "branch", "--contains", sha,
"--format=%(refname:short)"],
cwd=source_repo, capture=True,
)
containing = [b for b in r.stdout.strip().splitlines() if b]
if containing:
branch = _pick_preferred_branch(containing)
if branch_pushed.get(branch) != sha:
group.append(f"{sha}:refs/heads/{branch}")
branch_pushed[branch] = sha
push_groups.append(group)
elif refname.startswith("refs/heads/"):
branch = refname[len("refs/heads/"):]
if branch_pushed.get(branch) != sha:
push_groups.append([f"{sha}:refs/heads/{branch}"])
branch_pushed[branch] = sha
# else: already pushed to this hash via a tag push — skip
return push_groups
def check_lfs_endpoint(repo_dir: Path, extra_git_config: list = None):
"""Log the LFS endpoint that git-lfs selected for a repo (for debugging)."""
cwd = host_path_to_client(repo_dir)
cmd = ["git"]
if extra_git_config:
for kv in extra_git_config:
cmd += ["-c", kv]
cmd += ["lfs", "env"]
try:
r = client_exec(cmd, cwd=cwd, capture=True, check=False, timeout=10)
for line in r.stdout.splitlines():
if "Endpoint" in line or "SSH" in line:
log(f" LFS env: {line.strip()}")
except Exception:
pass
def _default_reclaim():
"""Reclaim page cache from core containers."""
reclaim_page_cache(CLIENT_CONTAINER)
reclaim_page_cache(GIT_SSH_CONTAINER)
def git_push_client(remote_url: str, source_repo: Path = None,
extra_git_config: list = None, extra_env: dict = None,
push_refspecs: list = None,
timeout: int = 3600,
container_cwd: str = None,
reclaim_fn=_default_reclaim,
trace_log: str = None,
stop_after: int = None) -> float:
"""Push source_repo to remote_url from inside the client container.
Strategy (when push_refspecs is provided):
Push one ref at a time. Each invocation creates a fresh SSH pool so the
pool-shutdown bug (pool dies after first batch in a multi-branch push)
never triggers. gc.auto=0 on the server repo prevents pack-file
accumulation from causing lock/temp-dir errors.
When None, falls back to --all for simple cases (synthetic dataset).
container_cwd: explicit path inside the container (e.g. /data-src for a
directly-mounted bare repo). When set, source_repo is ignored for path
resolution.
trace_log: when set, the *first* push invocation is run with GIT_TRACE=1,
GIT_TRANSFER_TRACE=1, and GIT_CURL_VERBOSE=1, and all output (stdout +
stderr) is saved to this path (on the host). Subsequent pushes (if using
push_refspecs) run without tracing.
"""
cwd = container_cwd if container_cwd else host_path_to_client(source_repo)
base_cmd = ["git",
# Suppress LFS lock-verify round-trip — server doesn't support it.
"-c", "lfs.locksverify=false",
"-c", f"lfs.concurrenttransfers={GIT_LFS_CONCURRENCY}"]
if extra_git_config:
for kv in extra_git_config:
base_cmd += ["-c", kv]
trace_env = None
if trace_log:
trace_env = {
"GIT_TRACE": "1",
"GIT_TRANSFER_TRACE": "1",
"GIT_CURL_VERBOSE": "1",
}
start = time.perf_counter()
if push_refspecs:
total = min(len(push_refspecs), stop_after) if stop_after else len(push_refspecs)
ndots = total // 10
if ndots > 0:
print("." * ndots, end="\r", flush=True)
for i, ref_group in enumerate(push_refspecs, 1):
if stop_after and i > stop_after:
break
# ref_group is a list of refspecs to push together.
if isinstance(ref_group, str):
ref_group = [ref_group] # backward compat
# Trace only the first push.
push_env = dict(extra_env) if extra_env else {}
if trace_env and i == 1:
push_env.update(trace_env)
r = client_exec(
base_cmd + ["push", remote_url] + ref_group,
cwd=cwd, env=push_env or None, timeout=timeout,
)
if trace_log and i == 1:
Path(trace_log).write_text(
f"=== TRACE: first push (refs={ref_group}) ===\n"
f"--- stdout ---\n{r.stdout}\n"
f"--- stderr ---\n{r.stderr}\n",
encoding="utf-8",
)
log(f" [trace] Saved first-push trace to {trace_log}")
if i % 10 == 0:
print("*", end="", flush=True)
if reclaim_fn and RECLAIM_INTERVAL > 0 and i % RECLAIM_INTERVAL == 0:
reclaim_fn()
if ndots > 0:
print(flush=True) # newline after progress
else:
push_env = dict(extra_env) if extra_env else {}
if trace_env:
push_env.update(trace_env)
r = client_exec(
base_cmd + ["push", "--all", remote_url],
cwd=cwd, env=push_env or None, timeout=timeout,
)
if trace_log:
Path(trace_log).write_text(
f"=== TRACE: push --all ===\n"
f"--- stdout ---\n{r.stdout}\n"
f"--- stderr ---\n{r.stderr}\n",
encoding="utf-8",
)
log(f" [trace] Saved push trace to {trace_log}")
return time.perf_counter() - start
def git_clone_client(remote_url: str, dest: Path,
extra_git_config: list = None, extra_env: dict = None,
timeout: int = 3600) -> float:
"""Clone remote_url into dest from inside the client container.
dest is a path on the host (inside WORK_DIR), accessible in the container
via the shared mount. The directory must not exist before cloning.
"""
if dest.exists():
shutil.rmtree(dest)
dest_client = host_path_to_client(dest)
cmd = ["git",
"-c", f"lfs.concurrenttransfers={GIT_LFS_CONCURRENCY}"]
if extra_git_config:
for kv in extra_git_config:
cmd += ["-c", kv]
cmd += ["clone", remote_url, dest_client]
start = time.perf_counter()
client_exec(cmd, cwd=WORK_DIR_IN_CLIENT, env=extra_env, timeout=timeout)
return time.perf_counter() - start
def git_lfs_pull_client(repo: Path, extra_git_config: list = None,
extra_env: dict = None,
extra_fetch_refspecs: list = None,
use_ssh_pull: bool = False,
timeout: int = 3600) -> float:
"""Fetch all LFS objects (all refs) and check out the working tree.
Fetches git refs/heads/* from origin so all branches are local, then
fetches all LFS objects with --all (traverses full history, not just tips).
use_ssh_pull: when True, use 'git lfs pull' per branch instead of
'git lfs fetch --all'. This forces the SSH transfer adapter which
is needed when the LFS server only supports SSH pkt-line transfer
(no HTTP batch API endpoint).
"""
cwd = host_path_to_client(repo)
base_cmd = ["git",
"-c", f"lfs.concurrenttransfers={GIT_LFS_CONCURRENCY}"]
if extra_git_config:
for kv in extra_git_config:
base_cmd += ["-c", kv]
start = time.perf_counter()
# Populate refs/heads/* so all branches are reachable (not just main).
client_exec(
base_cmd + ["fetch", "origin", "+refs/heads/*:refs/heads/*", "--update-head-ok"],
cwd=cwd, env=extra_env, timeout=timeout,
)
# Switch to the 'main' branch if it exists (Forgejo repos may default to
# 'master' which can be empty; the R0 content is on 'main').
client_exec(
["git", "checkout", "main"],
cwd=cwd, check=False, timeout=600,
)
if use_ssh_pull:
# Use 'git lfs pull' which goes through the SSH transfer adapter.
# Pull for the current branch (fetches LFS objects via pkt-line).
client_exec(
base_cmd + ["lfs", "pull"],
cwd=cwd, env=extra_env, timeout=timeout,
)
else:
# Fetch all LFS objects reachable from any ref (full history traversal).
# Uses the HTTP batch API.
client_exec(
base_cmd + ["lfs", "fetch", "--all", "origin"],
cwd=cwd, env=extra_env, timeout=timeout,
)
client_exec(base_cmd + ["lfs", "checkout"], cwd=cwd, env=extra_env, timeout=timeout)
return time.perf_counter() - start
def client_remote_url(repo_name: str) -> str:
"""SSH URL for a bare repo on git-ssh, reachable from inside the client container."""
return f"ssh://git@{GIT_SSH_INTERNAL}/home/git/{repo_name}.git"
# ---------------------------------------------------------------------------
# Binary building
# ---------------------------------------------------------------------------
def build_binaries():
log_step("Building binaries")
BINARIES_DIR.mkdir(exist_ok=True)
# Build desync git-lfs-transfer (from source, linux/amd64 for server container)
dest = BINARIES_DIR / "git-lfs-transfer"
log(f" Building desync git-lfs-transfer → {dest}")
run(
["go", "build", "-o", str(dest), "./cmd/git-lfs-transfer"],
cwd=DESYNC_SRC,
env={"CGO_ENABLED": "0", "GOOS": "linux", "GOARCH": "amd64"},
)
os.chmod(dest, 0o755)
log_ok("desync git-lfs-transfer built")
# Build desync git-lfs-desync (linux/amd64 for client container)
dest_agent = BINARIES_DIR / "git-lfs-desync"
log(f" Building desync git-lfs-desync → {dest_agent}")
run(
["go", "build", "-o", str(dest_agent), "./cmd/git-lfs-desync"],
cwd=DESYNC_SRC,
env={"CGO_ENABLED": "0", "GOOS": "linux", "GOARCH": "amd64"},
)
os.chmod(dest_agent, 0o755)
log_ok("desync git-lfs-desync built")
# Build the baseline (S1) git-lfs-transfer implementation.
if S1_LFS_TRANSFER == "scutiger":
# Rust, needs musl target for Alpine containers.
s1_dest = BINARIES_DIR / "s1-git-lfs-transfer"
if SCUTIGER_SRC.exists():
log(f" Building scutiger git-lfs-transfer → {s1_dest}")
run(
["cargo", "build", "--release", "--target", "x86_64-unknown-linux-musl", "-p", "scutiger-lfs"],
cwd=SCUTIGER_SRC, hide_output=True,
)
built = SCUTIGER_SRC / "target" / "x86_64-unknown-linux-musl" / "release" / "git-lfs-transfer"
if not built.exists():
raise RuntimeError(f"scutiger binary not found at {built}")
shutil.copy2(built, s1_dest)
os.chmod(s1_dest, 0o755)
log_ok("scutiger git-lfs-transfer built")
else:
log_warn(f"scutiger source not found at {SCUTIGER_SRC}, skipping build")
elif S1_LFS_TRANSFER == "charmbracelet":
s1_dest = BINARIES_DIR / "s1-git-lfs-transfer"
if CHARM_SRC.exists() and (CHARM_SRC / "go.mod").exists():
log(f" Building charmbracelet git-lfs-transfer → {s1_dest}")
run(
["go", "build", "-o", str(s1_dest), "."],
cwd=CHARM_SRC,
env={"CGO_ENABLED": "0", "GOOS": "linux", "GOARCH": "amd64"},
)
os.chmod(s1_dest, 0o755)
log_ok("charmbracelet git-lfs-transfer built")
else:
log_warn(f"charmbracelet source not found at {CHARM_SRC}, skipping build")
else:
log_warn(f"Unknown S1_LFS_TRANSFER={S1_LFS_TRANSFER!r}, skipping S1 build")
# Build custom git-lfs with transfer negotiation support.
lfs_dest = BINARIES_DIR / "git-lfs"
if GIT_LFS_SRC.exists() and (GIT_LFS_SRC / "go.mod").exists():
log(f" Building custom git-lfs → {lfs_dest}")
run(
["go", "build", "-o", str(lfs_dest), "."],
cwd=GIT_LFS_SRC,
env={"CGO_ENABLED": "0", "GOOS": "linux", "GOARCH": "amd64"},
)
os.chmod(lfs_dest, 0o755)
log_ok("custom git-lfs built")
elif not lfs_dest.exists():
log_warn(f"git-lfs source not found at {GIT_LFS_SRC} and no pre-built binary")
# Build the custom Forgejo Docker image if not already present.
r = run(["docker", "image", "inspect", "forgejo-desync:local"], hide_output=True, check=False)
if r.returncode != 0:
if FORGEJO_SRC.exists() and (FORGEJO_SRC / "Dockerfile").exists():
log(" Building forgejo-desync:local Docker image (this may take several minutes)...")
# Initialize Forgejo's nested desync submodule if needed.
run(["git", "submodule", "update", "--init"], cwd=FORGEJO_SRC, hide_output=True, check=False)
# The Makefile sets FORGEJO_VERSION from a VERSION file (git tags
# aren't available inside the Docker build context, and
# .dockerignore excludes /VERSION). Write it temporarily and
# remove the .dockerignore exclusion during the build.
version_file = FORGEJO_SRC / "VERSION"
dockerignore = FORGEJO_SRC / ".dockerignore"
created_version = not version_file.exists()
if created_version:
version = "0.0.0-dev"
r = run(["git", "describe", "--tags", "--always"], cwd=FORGEJO_SRC, capture=True, check=False)
if r.returncode == 0 and r.stdout.strip():
version = r.stdout.strip().lstrip("v")
version_file.write_text(version)
di_backup = None
if dockerignore.exists():
di_backup = dockerignore.read_text()
if "/VERSION" in di_backup:
dockerignore.write_text(di_backup.replace("/VERSION\n", "").replace("/VERSION", ""))
try:
run(
["docker", "build", "-t", "forgejo-desync:local", str(FORGEJO_SRC)],
timeout=1200,
)
finally:
if created_version:
version_file.unlink(missing_ok=True)
if di_backup is not None:
dockerignore.write_text(di_backup)
log_ok("forgejo-desync:local image built")
else:
log_warn(f"Forgejo source not found at {FORGEJO_SRC}, scenarios 8-12 will not work")
else:
log_ok("forgejo-desync:local image already exists")
# ---------------------------------------------------------------------------
# SSH key management
# ---------------------------------------------------------------------------
def setup_ssh_keys():
log_step("Setting up SSH keys")
KEYS_DIR.mkdir(exist_ok=True)
if not SSH_KEY.exists():
run(["ssh-keygen", "-t", "ed25519", "-f", str(SSH_KEY), "-N", "", "-q"])
log_ok(f"Generated SSH key pair at {SSH_KEY}")
else:
log_ok("SSH key pair already exists")
# ---------------------------------------------------------------------------
# Garage config generation
# ---------------------------------------------------------------------------
RPC_SECRET = None # generated once
def generate_garage_toml():
"""Generate a single garage.toml shared by all Garage instances."""
rpc_secret = subprocess.check_output(
["openssl", "rand", "-hex", "32"], text=True
).strip()
content = f"""metadata_dir = "/garage-meta"
data_dir = "/garage-data"
replication_factor = 1
rpc_bind_addr = "[::]:3901"
rpc_public_addr = "127.0.0.1:3901"
rpc_secret = "{rpc_secret}"
compression_level = "none"
block_ram_buffer_max = "512MiB"
block_max_concurrent_reads = 32
[s3_api]
s3_region = "garage"
api_bind_addr = "[::]:3900"
root_domain = ".s3.garage"
"""
GARAGE_TOML.write_text(content)
log_ok(f"Generated Garage config: {GARAGE_TOML.name}")
# ---------------------------------------------------------------------------
# Docker compose lifecycle
# ---------------------------------------------------------------------------
def start_services(work_dir: Path, lfs_dir: Path = None):
log_step("Starting Docker services")
generate_garage_toml()
BINARIES_DIR.mkdir(exist_ok=True)
KEYS_DIR.mkdir(exist_ok=True)
# Create placeholder binaries if they don't exist (for docker volume mounts)
for name in ["git-lfs-transfer", "s1-git-lfs-transfer", "git-lfs-desync", "git-lfs"]:
p = BINARIES_DIR / name
if not p.exists():
p.write_bytes(b"placeholder")
# Create placeholder authorized_keys
auth_keys = KEYS_DIR / "authorized_keys"
if not auth_keys.exists():
auth_keys.write_text("")
# Write .env so that manual `docker compose` invocations also get LFS_WORK_DIR.
env_lines = f"LFS_WORK_DIR={work_dir}\n"
compose_env = {"LFS_WORK_DIR": str(work_dir)}
if lfs_dir:
env_lines += f"LFS_LFS_DIR={lfs_dir}\n"
compose_env["LFS_LFS_DIR"] = str(lfs_dir)
(EXPERIMENT_DIR / ".env").write_text(env_lines)
compose(["up", "--build", "-d"], env=compose_env, timeout=300)
log_ok("Core services started (git-ssh + client)")
_wait_ssh_ready()
def _wait_ssh_ready():
"""Wait for SSH service to accept connections."""
log(" Waiting for SSH service...")
for _ in range(30):
try:
result = run(
["ssh", "-i", str(SSH_KEY),
"-o", "StrictHostKeyChecking=no",
"-o", "UserKnownHostsFile=/dev/null",
"-o", "ConnectTimeout=2",
"-o", "BatchMode=yes",
"-p", str(SSH_PORT),
"git@localhost", "echo ok"],
hide_output=True, check=False, timeout=5,
)
if result.returncode == 0:
log_ok("SSH ready")
return
except Exception:
pass
time.sleep(1)
log_warn("SSH may not be ready yet, continuing anyway")
# S3 services are started on demand per scenario to limit resource usage.
# Each service has a profile "s3" in docker-compose.yml and must be started
# explicitly. Track which services have been set up to avoid re-running
# idempotent-but-slow setup steps (Garage layout apply, bucket creation).
_s3_setup_done: set = set()
def _wait_s3_healthy(service: str, timeout_s: int = 30):
"""Wait until an S3 service responds to health/status checks."""
if service == "shared-minio":
for _ in range(timeout_s):
try:
urllib.request.urlopen(f"{SHARED_MINIO_ENDPOINT}/minio/health/live", timeout=2)
return
except Exception:
time.sleep(1)
log_warn(f"{service} health check timed out after {timeout_s}s")
return
if service == "shared-garage":
for _ in range(timeout_s):
try:
r = _shared_garage(["/garage", "status"])
if r.returncode == 0:
return
except Exception:
pass
time.sleep(1)
log_warn(f"{service} status check timed out after {timeout_s}s")
def start_s3_service(service: str, setup_fn=None):
"""Start an S3-profile service and run its setup function if needed."""
log(f" Starting {service}...")
compose(["--profile", "s3", "up", "-d", service], timeout=600)
_wait_s3_healthy(service)
if service not in _s3_setup_done and setup_fn:
setup_fn()
_s3_setup_done.add(service)
log_ok(f"{service} ready")
def stop_s3_service(service: str):
"""Stop an S3 service to free resources."""
log(f" Stopping {service}...")
compose(["--profile", "s3", "stop", service], timeout=120)
log_ok(f"{service} stopped")
def wipe_shared_minio():
"""Stop, remove, wipe volume, and restart the shared MinIO instance."""
log(" Wiping shared-minio volume...")
compose(["--profile", "s3", "stop", "shared-minio"], check=False, timeout=120)
compose(["--profile", "s3", "rm", "-f", "shared-minio"], check=False, timeout=120)
run(["docker", "volume", "rm", "-f", f"{COMPOSE_PROJECT}_shared-minio-data"],
hide_output=True, check=False, timeout=30)
_s3_setup_done.discard("shared-minio")
_container_full_ids.pop(SHARED_MINIO_CONTAINER, None)
log_ok("shared-minio volume wiped")
def wipe_shared_garage():
"""Stop, remove, wipe volumes, and restart the shared Garage instance."""
global SHARED_GARAGE_KEY_ID, SHARED_GARAGE_SECRET
log(" Wiping shared-garage volumes...")
compose(["--profile", "s3", "stop", "shared-garage"], check=False, timeout=120)
compose(["--profile", "s3", "rm", "-f", "shared-garage"], check=False, timeout=120)
run(["docker", "volume", "rm", "-f", f"{COMPOSE_PROJECT}_shared-garage-data"],
hide_output=True, check=False, timeout=30)
run(["docker", "volume", "rm", "-f", f"{COMPOSE_PROJECT}_shared-garage-meta"],
hide_output=True, check=False, timeout=30)
_s3_setup_done.discard("shared-garage")
_container_full_ids.pop(SHARED_GARAGE_CONTAINER, None)
SHARED_GARAGE_KEY_ID = None
SHARED_GARAGE_SECRET = None
log_ok("shared-garage volumes wiped")
_container_full_ids: dict = {}
def _get_container_full_id(container_name: str) -> str:
"""Return cached full container ID, refreshing on first call or after restart."""
if container_name not in _container_full_ids:
r = run(["docker", "inspect", container_name, "--format", "{{.Id}}"],
capture=True, check=False, timeout=10)
if r.returncode == 0:
_container_full_ids[container_name] = r.stdout.strip()
return _container_full_ids.get(container_name, "")
def reclaim_all_page_cache(*extra):
"""Reclaim page cache from core containers and any extras."""
for name in (CLIENT_CONTAINER, GIT_SSH_CONTAINER) + extra:
reclaim_page_cache(name)
def reclaim_page_cache(container_name: str):
"""Force kernel page cache reclaim for a container via cgroup v2 memory.reclaim.
Requires /sys/fs/cgroup mounted as /host-cgroup inside the container.
The write returns EAGAIN if less memory was reclaimable than requested,
which is expected and not an error.
"""
full_id = _get_container_full_id(container_name)
if not full_id:
return
run(["docker", "exec", "-u", "root", container_name,
"sh", "-c", f'echo 100G > /host-cgroup/docker/{full_id}/memory.reclaim'],
check=False, hide_output=True, timeout=10)
def start_core_containers():
"""Start (not recreate) the core containers and wait for SSH."""
log(" Starting core containers …")
compose(["start", "client", "git-ssh"], timeout=120)
_wait_ssh_ready()
def stop_all_containers():
"""Stop all running containers (preserving state and volumes)."""
compose(["--profile", "s3", "--profile", "forgejo", "stop"], check=False, timeout=120)
log_ok("Containers stopped")
def stop_services():
"""Remove all containers and volumes (final cleanup)."""
log_step("Stopping Docker services")
compose(["--profile", "s3", "--profile", "forgejo", "down"], check=False, timeout=300)
log_ok("Services stopped")
def write_client_desync_config():
"""Write a desync config file on the client with local-only settings.
Written to the default config location ($HOME/.config/desync/config.json)
so the agent picks it up automatically without --config flag.
This config provides client-side settings (cache, concurrency, resource
limits) that are NOT part of the server-advertised config. The server
provides store URLs, credentials, chunk-size, and digest via the SSH
transfer negotiation protocol; MergeServerConfig() combines the two.
"""
config = {