10620697748

Committed 29 Aug 2024 06:34PM UTC coverage: 31.923% (+0.01%) from 31.912%

Build # 10620697748

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #9663 from ForumMagnum/stream-llm-message-chunks

Stream llm message chunks

Run Details

5022 of 41078 branches covered (12.23%)

Branch coverage included in aggregate %.

23 of 81 new or added lines in 7 files covered. (28.4%)

4 existing lines in 3 files now uncovered.

29335 of 66547 relevant lines covered (44.08%)

188.28 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.26

/packages/lesswrong/server/updateScores.ts

import { getCollection, Vulcan } from "./vulcan-lib";
import {
  TIME_DECAY_FACTOR,
  getSubforumScoreBoost,
  SCORE_BIAS,
} from '../lib/scoring';
import { runSqlQuery } from "@/server/sql/sqlClient";
import chunk from "lodash/chunk";
import compact from "lodash/compact";

// INACTIVITY_THRESHOLD_DAYS =  number of days after which a single vote will not have a big enough effect to trigger a score update
//      and posts can become inactive
const INACTIVITY_THRESHOLD_DAYS = 30;

interface BatchUpdateParams {
  inactive?: boolean;
  forceUpdate?: boolean;
}

const getPgCollectionProjections = (collectionName: VoteableCollectionName) => {
  const proj = {
    _id: '"_id"',
    postedAt: '"postedAt"',
    scoreDate: '"postedAt" AS "scoreDate"',
    score: '"score"',
    baseScore: '"baseScore"',
  };
  switch (collectionName) {
    case "Posts":
      proj.scoreDate = `(CASE WHEN "frontpageDate" IS NULL
        THEN "postedAt"
        ELSE "frontpageDate" END) AS "scoreDate"`;
      proj.baseScore = `("baseScore" +
        (CASE WHEN "frontpageDate" IS NULL THEN 0 ELSE 10 END) +
        (CASE WHEN "curatedDate" IS NULL THEN 0 ELSE 10 END)) AS "baseScore"`;
      break;
    case "Comments":
      const {base, magnitude, duration, exponent} = getSubforumScoreBoost();
      const ageHours = '(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP - "createdAt") / 3600)';
      proj.baseScore = `("baseScore" + (CASE WHEN "tagCommentType" = 'SUBFORUM'
        THEN GREATEST(
          ${base},
          ${magnitude} * (1 - POW(${ageHours} / ${duration}, ${exponent}))
        )
        ELSE 0 END)) as "baseScore"`;
      break;
    default:
      break;
  }
  return Object.values(proj);
}

const getBatchItemsPg = async <N extends VoteableCollectionName>(collection: CollectionBase<N>, inactive: boolean, forceUpdate: boolean) => {
  const {collectionName} = collection;
  if (!["Posts", "Comments"].includes(collectionName)) {
    return [];
  }

  const ageHours = 'EXTRACT(EPOCH FROM CURRENT_TIMESTAMP - "postedAt") / 3600';
  return runSqlQuery(`
    -- updateScores.getBatchItemsPg
    SELECT
      q.*,
      ns."newScore",
      q."inactive" AS "inactive",
      ABS("score" - ns."newScore") > ns."singleVotePower" AS "scoreDiffSignificant",
      ns."singleVotePower" AS "singleVotePower",
      (${ageHours}) > ($1 * 24) AS "oldEnough"
    FROM (
      SELECT ${getPgCollectionProjections(collectionName).join(", ")}
        , "${collectionName}".inactive as inactive
      FROM "${collectionName}"
      WHERE
        "postedAt" < CURRENT_TIMESTAMP AND
        ${inactive ? '"inactive" = TRUE' : '("inactive" = FALSE OR "inactive" IS NULL)'}
    ) q, LATERAL (SELECT
      "baseScore" / POW(${ageHours} + $2, $3) AS "newScore",
      1.0 / POW(${ageHours} + $2, $3) AS "singleVotePower"
    ) ns
    ${forceUpdate ? "" : 'WHERE ABS("score" - ns."newScore") > ns."singleVotePower" OR NOT q."inactive"'}
  `, [INACTIVITY_THRESHOLD_DAYS, SCORE_BIAS, TIME_DECAY_FACTOR.get()], "read");
}

const getBatchItems = <N extends VoteableCollectionName>(collection: CollectionBase<N>, inactive: boolean, forceUpdate: boolean) => {
  return getBatchItemsPg(collection, inactive, forceUpdate)
}

export const batchUpdateScore = async ({collection, inactive = false, forceUpdate = false}: BatchUpdateParams & { collection: CollectionBase<VoteableCollectionName> }) => {
  const items = await getBatchItems(collection, inactive, forceUpdate);
  let updatedDocumentsCounter = 0;

  const batches = chunk(items, 1000); // divide items into chunks of 1000

  for (let batch of batches) {
    let itemUpdates = compact(batch.map((i: AnyBecauseTodo) => {
      if (forceUpdate || i.scoreDiffSignificant) {
        updatedDocumentsCounter++;
        return {
          updateOne: {
            filter: {_id: i._id},
            update: {$set: {score: i.newScore, inactive: false}},
            upsert: false,
          }
        }
      } else if (i.oldEnough && !i.inactive) {
        // only set a post as inactive if it's older than n days
        return {
          updateOne: {
            filter: {_id: i._id},
            update: {$set: {inactive: true}},
            upsert: false,
          }
        }
      }
    }));

    if (itemUpdates && itemUpdates.length) {
      await collection.rawCollection().bulkWrite(itemUpdates, {ordered: false});
    }
  }
  return updatedDocumentsCounter;
}

export const batchUpdateScoreByName = ({collectionName, inactive = false, forceUpdate = false}: BatchUpdateParams & { collectionName: VoteableCollectionName }) => {
  const collection = getCollection(collectionName);
  return batchUpdateScore({collection, inactive, forceUpdate});
}

Vulcan.batchUpdateScoreByName = batchUpdateScoreByName;

1	import { getCollection, Vulcan } from "./vulcan-lib";	155✔
2	import {
3	TIME_DECAY_FACTOR,
4	getSubforumScoreBoost,
5	SCORE_BIAS,
6	} from '../lib/scoring';	18✔
7	import { runSqlQuery } from "@/server/sql/sqlClient";	18✔
8	import chunk from "lodash/chunk";	18✔
9	import compact from "lodash/compact";	18✔
10
11	// INACTIVITY_THRESHOLD_DAYS = number of days after which a single vote will not have a big enough effect to trigger a score update
12	// and posts can become inactive
13	const INACTIVITY_THRESHOLD_DAYS = 30;	18✔
14
15	interface BatchUpdateParams {
16	inactive?: boolean;
17	forceUpdate?: boolean;
18	}
19
20	const getPgCollectionProjections = (collectionName: VoteableCollectionName) => {	18✔
21	const proj = {	101✔
22	_id: '"_id"',
23	postedAt: '"postedAt"',
24	scoreDate: '"postedAt" AS "scoreDate"',
25	score: '"score"',
26	baseScore: '"baseScore"',
27	};
28	switch (collectionName) {	101!
29	case "Posts":
30	proj.scoreDate = `(CASE WHEN "frontpageDate" IS NULL	101✔
31	THEN "postedAt"
32	ELSE "frontpageDate" END) AS "scoreDate"`;
33	proj.baseScore = `("baseScore" +	101✔
34	(CASE WHEN "frontpageDate" IS NULL THEN 0 ELSE 10 END) +
35	(CASE WHEN "curatedDate" IS NULL THEN 0 ELSE 10 END)) AS "baseScore"`;
36	break;	101✔
37	case "Comments":
38	const {base, magnitude, duration, exponent} = getSubforumScoreBoost();	×
39	const ageHours = '(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP - "createdAt") / 3600)';	×
40	proj.baseScore = `("baseScore" + (CASE WHEN "tagCommentType" = 'SUBFORUM'	×
41	THEN GREATEST(
42	${base},
43	${magnitude} * (1 - POW(${ageHours} / ${duration}, ${exponent}))
44	)
45	ELSE 0 END)) as "baseScore"`;
46	break;	×
47	default:
NEW 48	break;	×
49	}
50	return Object.values(proj);	101✔
51	}
52
53	const getBatchItemsPg = async <N extends VoteableCollectionName>(collection: CollectionBase<N>, inactive: boolean, forceUpdate: boolean) => {	18✔
54	const {collectionName} = collection;	101✔
55	if (!["Posts", "Comments"].includes(collectionName)) {	101!
56	return [];	×
57	}
58
59	const ageHours = 'EXTRACT(EPOCH FROM CURRENT_TIMESTAMP - "postedAt") / 3600';	101✔
60	return runSqlQuery(`	101✔
61	-- updateScores.getBatchItemsPg
62	SELECT
63	q.*,
64	ns."newScore",
65	q."inactive" AS "inactive",
66	ABS("score" - ns."newScore") > ns."singleVotePower" AS "scoreDiffSignificant",
67	ns."singleVotePower" AS "singleVotePower",
68	(${ageHours}) > ($1 * 24) AS "oldEnough"
69	FROM (
70	SELECT ${getPgCollectionProjections(collectionName).join(", ")}
71	, "${collectionName}".inactive as inactive
72	FROM "${collectionName}"
73	WHERE
74	"postedAt" < CURRENT_TIMESTAMP AND
75	${inactive ? '"inactive" = TRUE' : '("inactive" = FALSE OR "inactive" IS NULL)'}	101!
76	) q, LATERAL (SELECT
77	"baseScore" / POW(${ageHours} + $2, $3) AS "newScore",
78	1.0 / POW(${ageHours} + $2, $3) AS "singleVotePower"
79	) ns
80	${forceUpdate ? "" : 'WHERE ABS("score" - ns."newScore") > ns."singleVotePower" OR NOT q."inactive"'}	101!
81	`, [INACTIVITY_THRESHOLD_DAYS, SCORE_BIAS, TIME_DECAY_FACTOR.get()], "read");
82	}
83
84	const getBatchItems = <N extends VoteableCollectionName>(collection: CollectionBase<N>, inactive: boolean, forceUpdate: boolean) => {	18✔
85	return getBatchItemsPg(collection, inactive, forceUpdate)	101✔
86	}
87
88	export const batchUpdateScore = async ({collection, inactive = false, forceUpdate = false}: BatchUpdateParams & { collection: CollectionBase<VoteableCollectionName> }) => {	18✔
89	const items = await getBatchItems(collection, inactive, forceUpdate);	101✔
90	let updatedDocumentsCounter = 0;	101✔
91
92	const batches = chunk(items, 1000); // divide items into chunks of 1000	101✔
93
94	for (let batch of batches) {	101✔
95	let itemUpdates = compact(batch.map((i: AnyBecauseTodo) => {	100✔
96	if (forceUpdate \|\| i.scoreDiffSignificant) {	1,171✔
97	updatedDocumentsCounter++;	97✔
98	return {	97✔
99	updateOne: {
100	filter: {_id: i._id},
101	update: {$set: {score: i.newScore, inactive: false}},
102	upsert: false,
103	}
104	}
105	} else if (i.oldEnough && !i.inactive) {	1,074✔
106	// only set a post as inactive if it's older than n days
107	return {	1✔
108	updateOne: {
109	filter: {_id: i._id},
110	update: {$set: {inactive: true}},
111	upsert: false,
112	}
113	}
114	}
115	}));
116
117	if (itemUpdates && itemUpdates.length) {	100✔
118	await collection.rawCollection().bulkWrite(itemUpdates, {ordered: false});	95✔
119	}
120	}
121	return updatedDocumentsCounter;	101✔
122	}
123
124	export const batchUpdateScoreByName = ({collectionName, inactive = false, forceUpdate = false}: BatchUpdateParams & { collectionName: VoteableCollectionName }) => {	18!
125	const collection = getCollection(collectionName);	×
126	return batchUpdateScore({collection, inactive, forceUpdate});	×
127	}
128
129	Vulcan.batchUpdateScoreByName = batchUpdateScoreByName;	18✔

ForumMagnum / ForumMagnum / 10620697748

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous