Merge remote-tracking branch 'origin/master' into explorer-merge-search-terminal

Author: haraldschilly

src/package.json

@@ -25,6 +25,7 @@
     "nats-server-verbose": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/install').main()\" && node -e \"require('@cocalc/backend/nats/conf').main()\" && node -e \"require('@cocalc/backend/nats/server').main({verbose:true})\"",
     "nats-cli": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/cli').main()\"",
     "nats-sys": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/cli').main({user:'sys'})\"",
+    "nats-tiered-storage": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/server && DEBUG=cocalc:* DEBUG_CONSOLE=yes node -e \"require('@cocalc/server/nats/tiered-storage').init()\"",
     "local-ci": "./scripts/ci.sh"
   },
   "repository": {

src/packages/backend/nats/cli.ts

@@ -10,11 +10,14 @@ import { join } from "path";
 import { spawnSync } from "node:child_process";
 import { natsServerUrl } from "./conf";
 
+const natsBin = join(data, "nats", "bin");
+
 export function natsCoCalcUserEnv({ user = natsUser }: { user?: string } = {}) {
   return {
     NATS_URL: natsServerUrl,
     NATS_PASSWORD: natsPassword,
     NATS_USER: user ?? natsUser,
+    PATH: `${natsBin}:${process.env.PATH}`,
   };
 }
 
@@ -37,17 +40,16 @@ function params({ user }) {
 
 export function main({ user = natsUser }: { user?: string } = {}) {
   let { command, args, env } = params({ user });
-  const PATH0 = join(data, "nats", "bin");
   console.log("# Use CoCalc config of NATS (nats and nsc) via this subshell:");
   console.log(
     JSON.stringify(
-      { ...env, NATS_PASSWORD: "xxx", PATH: PATH0 + ":..." },
+      { ...env, NATS_PASSWORD: "xxx", PATH: natsBin + ":..." },
       undefined,
       2,
     ),
   );
   spawnSync(command, args, {
-    env: { ...env, PATH: `${PATH0}:${process.env.PATH}` },
+    env: { ...env, PATH: `${natsBin}:${process.env.PATH}` },
     stdio: "inherit",
   });
 }

src/packages/frontend/client/console.ts

@@ -45,6 +45,7 @@ export function setup_global_cocalc(client): void {
 
   const cocalc: any = window.cc ?? {};
   cocalc.client = client;
+  cocalc.nats = client.nats_client;
   cocalc.misc = require("@cocalc/util/misc");
   cocalc.immutable = require("immutable");
   cocalc.done = cocalc.misc.done;

src/packages/frontend/cspell.json

@@ -54,7 +54,8 @@
     "isdir",
     "mintime",
     "PoweroffOutlined",
-    "immutablejs"
+    "immutablejs",
+    "reuseinflight"
   ],
   "flagWords": [],
   "ignorePaths": ["node_modules/**", "dist/**", "dist-ts/**", "build/**"],

src/packages/frontend/customize.tsx

@@ -177,6 +177,7 @@ export interface CustomizeState {
   selectable_llms: List<string>;
   default_llm?: string;
   user_defined_llm: boolean;
+  llm_default_quota?: number;
 
   insecure_test_mode?: boolean;
 
@@ -334,7 +335,7 @@ function process_customize(obj) {
   for (const k in site_settings_conf) {
     const v = site_settings_conf[k];
     obj[k] =
-      obj[k] != null ? obj[k] : (v.to_val?.(v.default, obj_orig) ?? v.default);
+      obj[k] != null ? obj[k] : v.to_val?.(v.default, obj_orig) ?? v.default;
   }
   // the llm markup special case
   obj.llm_markup = obj_orig._llm_markup ?? 30;

src/packages/frontend/frame-editors/latex-editor/actions.ts

@@ -664,16 +664,24 @@ export class Actions extends BaseActions<LatexEditorState> {
   }
 
   public async explicit_save() {
-    const account: any = this.redux.getStore("account");
+    const account = this.redux.getStore("account");
     if (
-      account == null ||
-      !account.getIn(["editor_settings", "build_on_save"]) ||
+      !account?.getIn(["editor_settings", "build_on_save"]) ||
       !this.is_likely_master()
     ) {
-      await this.save_all(true);
+      // kicks off a save of all relevant files
+      // Obviously, do not make this save_all(true), because
+      // that would end up calling this very function again
+      // crashing the browser in an INFINITE RECURSSION
+      // (this was a bug for a while!).
+      // Also, the save of the related files is NOT
+      // explicit -- the user is only explicitly saving this
+      // file.  Explicit save is mainly about deleting trailing
+      // whitespace and launching builds.
+      await this.save_all(false);
       return;
     }
-    await this.build(); // kicks off a save of all relevant files
+    await this.build();
   }
 
   // used by generic framework – this is bound to the instance, otherwise "this" is undefined, hence

src/packages/frontend/frame-editors/terminal-editor/nats-terminal-connection.ts

@@ -236,7 +236,7 @@ export class NatsTerminalConnection extends EventEmitter {
   private setReady = async () => {
     // wait until after render loop of terminal before allowing writing,
     // or we get corruption.
-    await delay(1);
+    await delay(250);
     this.setState("running");
     this.emit("ready");
   };

src/packages/frontend/nats/client.ts

@@ -60,6 +60,7 @@ import Cookies from "js-cookie";
 import { ACCOUNT_ID_COOKIE } from "@cocalc/frontend/client/client";
 import { isConnected, waitUntilConnected } from "@cocalc/nats/util";
 import { info as refCacheInfo } from "@cocalc/util/refcache";
+import * as tieredStorage from "@cocalc/nats/tiered-storage/client";
 
 const NATS_STATS_INTERVAL = 2500;
 
@@ -673,6 +674,8 @@ export class NatsClient extends EventEmitter {
   waitUntilConnected = async () => await waitUntilConnected();
 
   refCacheInfo = () => refCacheInfo();
+
+  tieredStorage = tieredStorage;
 }
 
 function setDeleted({ project_id, path, deleted }) {

src/packages/frontend/project/explorer/file-listing/no-files.tsx

@@ -106,6 +106,7 @@ export default function NoFiles({
           margin: "0 auto",
           height: "80px",
           fontSize: "24px",
+          padding: "30px",
         }}
         onClick={(): void => {
           if (file_search.length === 0) {

src/packages/frontend/purchases/all-quotas-config.tsx

@@ -53,6 +53,9 @@ export default function AllQuotasConfig() {
   const lastFetchedQuotasRef = useRef<ServiceQuota[] | null>(null);
   const [changed, setChanged] = useState<boolean>(false);
   const selectableLLMs = useTypedRedux("customize", "selectable_llms");
+  // The 10 is just for the initial rollout, the value is customizable
+  const llm_default_quota =
+    useTypedRedux("customize", "llm_default_quota") ?? 10;
 
   const getQuotas = async () => {
     let quotas, charges;
@@ -74,17 +77,19 @@ export default function AllQuotasConfig() {
       const spec = QUOTA_SPEC[service];
       if (spec.noSet) continue;
       const llmModel = service2model_core(service);
-      if (llmModel != null) {
+      const isLLM = llmModel != null;
+      if (isLLM) {
         // We do not show those models, which can't be selected by users OR are free in the first place
         const cost = LLM_COST[llmModel];
         if (!selectableLLMs.includes(llmModel) || cost?.free === true) {
           continue;
         }
       }
+      const defaultQuota: number = isLLM ? llm_default_quota : 0;
       w[service] = {
         current: charges[service] ?? 0,
         service: service as Service,
-        quota: services[service] ?? 0,
+        quota: services[service] ?? defaultQuota,
       };
     }
     try {
@@ -154,8 +159,8 @@ export default function AllQuotasConfig() {
       dataIndex: "quota",
       align: "center" as "center",
       render: (quota: number, _record: ServiceQuota, index: number) => {
-        const presets =
-          QUOTA_SPEC[_record.service]?.category === "ai" ? PRESETS_LLM : PRESETS;
+        const isLLM = QUOTA_SPEC[_record.service]?.category === "ai";
+        const presets = isLLM ? PRESETS_LLM : PRESETS;
 
         return (
           <Dropdown

src/packages/hub/hub.ts

@@ -44,6 +44,7 @@ import initExpressApp from "./servers/express-app";
 import {
   loadNatsConfiguration,
   initNatsDatabaseServer,
+  initNatsTieredStorage,
   initNatsServer,
 } from "@cocalc/server/nats";
 import initHttpRedirect from "./servers/http-redirect";
@@ -194,6 +195,11 @@ async function startServer(): Promise<void> {
   if (program.natsDatabaseServer) {
     await initNatsDatabaseServer();
   }
+  if (program.natsTieredStorage) {
+    // currently there must be exactly ONE of these, running on the same
+    // node as the nats-server.  E.g., for development it's just part of the server.
+    await initNatsTieredStorage();
+  }
 
   if (program.websocketServer) {
     // Initialize the version server -- must happen after updating schema
@@ -525,6 +531,7 @@ async function main(): Promise<void> {
     program.websocketServer =
       program.natsServer =
       program.natsDatabaseServer =
+      program.natsTieredStorage =
       program.proxyServer =
       program.nextServer =
       program.mentions =

src/packages/nats/sync/general-kv.ts

@@ -127,6 +127,9 @@ import { headers as createHeaders } from "@nats-io/nats-core";
 import type { MsgHdrs } from "@nats-io/nats-core";
 import type { ValueType } from "@cocalc/nats/types";
 import { isConnected, waitUntilConnected } from "@cocalc/nats/util";
+import { ENFORCE_LIMITS_THROTTLE_MS } from "./stream";
+import { asyncDebounce } from "@cocalc/util/async-utils";
+import { waitUntilReady } from "@cocalc/nats/tiered-storage/client";
 
 const PUBLISH_TIMEOUT = 15000;
 
@@ -143,10 +146,6 @@ const CONNECTION_CHECK_INTERVAL = 5000;
 // which is convenient for consistency.  This is not consistent with NATS's
 // own KV store limit naming.
 
-// Significant throttling is VERY, VERY important, since purging old messages frequently
-// seems to put a very significant load on NATS!
-const ENFORCE_LIMITS_THROTTLE_MS = process.env.COCALC_TEST_MODE ? 100 : 30000;
-
 export interface KVLimits {
   // How many keys may be in the KV store. Oldest keys will be removed
   // if the key-value store exceeds this size. -1 for unlimited.
@@ -231,6 +230,7 @@ export class GeneralKV<T = any> extends EventEmitter {
     if (this.all != null) {
       return;
     }
+    await waitUntilReady(this.name);
     const kvm = new Kvm(this.env.nc);
     await waitUntilConnected();
     this.kv = await kvm.create(this.name, {
@@ -355,6 +355,9 @@ export class GeneralKV<T = any> extends EventEmitter {
   };
 
   private restartWatch = () => {
+    // this triggers the end of the "for await (const x of this.watch) {"
+    // loop in startWatch, which results in another watch starting,
+    // assuming the object isn't closed.
     this.watch?.stop();
   };
 
@@ -499,7 +502,16 @@ export class GeneralKV<T = any> extends EventEmitter {
   };
 
   private monitorWatch = async () => {
-    this.env.nc.on?.("reconnect", this.restartWatch);
+    if (this.env.nc.on != null) {
+      this.env.nc.on("reconnect", this.restartWatch);
+      this.env.nc.on("status", ({ type }) => {
+        if (type == "reconnect") {
+          this.ensureWatchIsValid();
+        }
+      });
+    } else {
+      this.checkWatchOnReconnect();
+    }
     while (this.revisions != null) {
       if (!(await isConnected())) {
         await waitUntilConnected();
@@ -511,6 +523,52 @@ export class GeneralKV<T = any> extends EventEmitter {
     }
   };
 
+  private ensureWatchIsValid = asyncDebounce(
+    async () => {
+      await waitUntilConnected();
+      await delay(2000);
+      const isValid = await this.isWatchStillValid();
+      if (!isValid) {
+        if (this.kv == null) {
+          return;
+        }
+        console.log(`nats kv: ${this.name} -- watch not valid, so recreating`);
+        await this.restartWatch();
+      }
+    },
+    3000,
+    { leading: false, trailing: true },
+  );
+
+  private isWatchStillValid = async () => {
+    await waitUntilConnected();
+    if (this.kv == null || this.watch == null) {
+      return false;
+    }
+    try {
+      await this.watch._data.info();
+      return true;
+    } catch (err) {
+      console.log(`nats: watch info error -- ${err}`);
+      return false;
+    }
+  };
+
+  private checkWatchOnReconnect = async () => {
+    while (this.kv != null) {
+      try {
+        for await (const { type } of await this.env.nc.status()) {
+          if (type == "reconnect") {
+            await this.ensureWatchIsValid();
+          }
+        }
+      } catch {
+        await delay(15000);
+        await this.ensureWatchIsValid();
+      }
+    }
+  };
+
   close = () => {
     if (this.revisions == null) {
       // already closed