File size: 4,533 Bytes
34fef0e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7156a50
34fef0e
 
 
 
 
 
 
 
 
1ae42b9
34fef0e
1ae42b9
34fef0e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
ca78461
34fef0e
 
 
 
 
 
 
 
 
 
 
fc9f257
34fef0e
 
 
 
 
 
 
 
 
 
1ae42b9
34fef0e
 
 
 
 
1ae42b9
34fef0e
 
 
 
 
 
 
1ae42b9
 
 
34fef0e
1ae42b9
34fef0e
 
 
 
 
7156a50
 
1ae42b9
 
 
 
7156a50
 
34fef0e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1ae42b9
34fef0e
1ae42b9
 
 
34fef0e
 
1ae42b9
34fef0e
 
 
 
 
 
 
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
import { existsSync } from "node:fs";
import type { OperatorAcl } from "@harbor-hf/contracts";
import { deterministicId, sha256 } from "@harbor-hf/contracts";
import {
  ControlService,
  FilesystemObjectStore,
  type ImmutableObjectStore,
  loadBuiltInProfiles,
  Projection,
  Reconciler,
  ResultPublisher,
} from "@harbor-hf/control-core";
import {
  attestInferenceToken,
  HuggingFaceActions,
  HuggingFaceBucketStore,
  NoopActions,
} from "@harbor-hf/hf-adapters";
import { AuthenticationService, AuthStore } from "./auth.js";
import type { AppConfig } from "./config.js";

export interface Runtime {
  config: AppConfig;
  projection: Projection;
  store: ImmutableObjectStore;
  service: ControlService;
  auth: AuthenticationService;
  reconciler: Reconciler;
  readonly ready: boolean;
  initialize(): Promise<void>;
  start(onReconcilerError?: (error: unknown) => void): void;
  close(): Promise<void>;
}

export async function createRuntime(config: AppConfig): Promise<Runtime> {
  if (config.store_mode === "filesystem" && !existsSync(config.bucket_root))
    throw new Error("filesystem object-store root is missing");
  const store: ImmutableObjectStore =
    config.store_mode === "bucket"
      ? new HuggingFaceBucketStore({
          bucketId: config.bucket_id,
          accessToken: config.hf_token ?? "",
        })
      : new FilesystemObjectStore(config.bucket_root);
  const projection = await Projection.open(config.projection_path);
  const profiles = await loadBuiltInProfiles(config.profiles_root);
  const service = new ControlService(config.namespace, store, projection, profiles);
  service.configureCapacityProfile(config.capacity_profile_alias);
  const authStore = await AuthStore.open(config.auth_path);
  const auth = new AuthenticationService(
    config.auth_mode,
    authStore,
    config.oauth,
    () => projection.latestAcl(),
  );
  const hfActions = config.hf_token
    ? new HuggingFaceActions({
        namespace: config.namespace,
        accessToken: config.hf_token,
        taskImageMirrorRepository: config.task_image_mirror_repository,
        ...(config.hf_inference_token
          ? { inferenceToken: config.hf_inference_token }
          : {}),
        controlUrl: config.public_origin,
      })
    : null;
  const external = hfActions ?? new NoopActions();
  const publisher = new ResultPublisher(store, projection, service);
  const reconciler = new Reconciler(service, projection, external, publisher, {
    interval_ms: config.reconcile_interval_ms,
    sync_interval_ms: config.sync_interval_ms,
    observation_interval_ms: config.observe_interval_ms,
    worker_receipt_grace_ms: config.worker_receipt_grace_ms,
    batch_size: 16,
  });
  const abort = new AbortController();
  let initializationReady = false;
  return {
    config,
    projection,
    store,
    service,
    auth,
    reconciler,
    get ready() {
      return initializationReady && projection.system().ready;
    },
    async initialize() {
      initializationReady = false;
      if (config.hf_inference_token)
        await attestInferenceToken({ accessToken: config.hf_inference_token });
      await auth.initialize();
      await projection.rebuild(store);
      await service.initialize(profiles);
      if (config.write_mode !== "disabled") {
        if (!service.capacityProfileOrNull())
          await service.setMaxActiveJobs(
            config.max_active_jobs,
            `capacity-bootstrap-${config.max_active_jobs}`,
          );
        service.requireCapacityProfile();
      }
      if (
        !(await projection.latestAcl()) &&
        config.bootstrap_operator_subjects.length > 0
      ) {
        const operators = [...new Set(config.bootstrap_operator_subjects)].sort();
        const acl: OperatorAcl = {
          schema_version: "v1",
          kind: "operator.acl",
          record_id: deterministicId("operator-acl", sha256(operators.join("\u0000"))),
          created_at: new Date().toISOString(),
          actor: { subject: "harbor-hf-bootstrap", role: "migration" },
          operators,
          readers: [],
        };
        await service.append(acl);
      }
      initializationReady = true;
    },
    start(onReconcilerError?: (error: unknown) => void) {
      if (config.write_mode !== "disabled")
        reconciler.start(abort.signal, onReconcilerError);
    },
    async close() {
      initializationReady = false;
      abort.abort();
      await reconciler.stop();
      authStore.close();
      await projection.close();
    },
  };
}