Added config param T_LIMIT_NUMBER_OF_CHANGES

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## v3.26.1
 
+__NEW Features__
+- New configuratiuon parameter: `T_LIMIT_NUMBER_OF_CHANGES` - Number of change docs from the Couchdb changes feed queried 
+  by reporting-worker (i.e. use as the limit parameter). Default: 200.
+
 __Fixes__
 - Student subtest report incorrect for custom logic inputs [#3464](https://github.com/Tangerine-Community/Tangerine/issues/3464)
 - Init paid-worker file when server restarted. 

config.defaults.sh

@@ -59,6 +59,9 @@ T_PAID_ALLOWANCE="unlimited"
 # Reporting delay determines how quickly uploads will get processed and show up in reporting outputs such as CSV. Time is in milliseconds and default is 5 minutes.
 T_REPORTING_DELAY="300000"
 
+# Number of change docs from the Couchdb changes feed queried by reporting-worker (i.e. use as the limit parameter)
+T_LIMIT_NUMBER_OF_CHANGES=200
+
 # Limit processing to certain group dbs.
 T_ONLY_PROCESS_THESE_GROUPS=""
 

develop-tangy-form-libs.sh

@@ -190,6 +190,8 @@ OPTIONS="--link $T_COUCHDB_CONTAINER_NAME:couchdb \
   --env \"T_PASSWORD_RECIPE=$T_PASSWORD_RECIPE\" \
   --env \"T_JWT_ISSUER=$T_JWT_ISSUER\" \
   --env \"T_JWT_EXPIRES_IN=$T_JWT_EXPIRES_IN\" \
+  --env \"T_ONLY_PROCESS_THESE_GROUPS=$T_ONLY_PROCESS_THESE_GROUPS\" \
+  --env \"T_LIMIT_NUMBER_OF_CHANGES=$T_LIMIT_NUMBER_OF_CHANGES\" \
   $T_PORT_MAPPING \
   -p 9229:9229 \
   -p 9228:9228 \

develop.sh

@@ -197,6 +197,7 @@ OPTIONS="--link $T_COUCHDB_CONTAINER_NAME:couchdb \
   --env \"T_JWT_ISSUER=$T_JWT_ISSUER\" \
   --env \"T_JWT_EXPIRES_IN=$T_JWT_EXPIRES_IN\" \
   --env \"T_ONLY_PROCESS_THESE_GROUPS=$T_ONLY_PROCESS_THESE_GROUPS\" \
+  --env \"T_LIMIT_NUMBER_OF_CHANGES=$T_LIMIT_NUMBER_OF_CHANGES\" \
   $T_PORT_MAPPING \
   -p 9229:9229 \
   -p 9228:9228 \

server/src/reporting/reporting-worker.js

@@ -127,6 +127,7 @@ async function addGroup(group) {
   log.info(`Add group to reporting worker state: ${group._id}`)
   // Something may have paused the process like clearing cache.
   while (await isPaused()) {
+    log.info(`Pausing reporting-worker for ${REPORTING_WORKER_PAUSE_LENGTH} ms.`)
     await sleep(REPORTING_WORKER_PAUSE_LENGTH)
   }
   await setWorkingFlag()
@@ -143,6 +144,7 @@ async function addGroup(group) {
  */
 
 async function batch() {
+  let workerState;
   try {
     // Something may have paused the process like clearing cache.
     while (await isPaused()) {
@@ -151,7 +153,10 @@ async function batch() {
     await setWorkingFlag()
     // Now it's safe to get the state.
     workerState = await getWorkerState()
-    workerState = Object.assign({} , defaultState, workerState)
+    workerState = Object.assign({}, defaultState, workerState)
+    if (process.env.T_LIMIT_NUMBER_OF_CHANGES) {
+      workerState.batchSizePerDatabase = process.env.T_LIMIT_NUMBER_OF_CHANGES
+    }
     const DB = PouchDB.defaults(workerState.pouchDbDefaults)
     const startTime = new Date().toISOString()
     let processed = 0
@@ -163,7 +168,7 @@ async function batch() {
       // log.info('onlyProcessTheseGroups from T_ONLY_PROCESS_THESE_GROUPS: ' + onlyProcessTheseGroups)
     }
     // Process batch.
-    for (let database of workerState.databases) { 
+    for (let database of workerState.databases) {
       let processGroup = false
       if (onlyProcessTheseGroups.length === 0 || onlyProcessTheseGroups.includes(database.name)) {
         processGroup = true
@@ -178,7 +183,7 @@ async function batch() {
           include_docs: false
         })
         if (changes.results.length > 0) {
-          log.debug("Processing a batch from change seq: " + database.sequence + " with a batchSizePerDatabase of " + workerState.batchSizePerDatabase)
+          log.debug("Processing a batch of " + changes.results.length + " changes from seq: " + database.sequence + " with a batchSizePerDatabase of " + workerState.batchSizePerDatabase)
           for (let change of changes.results) {
             try {
               await changeProcessor(change, db)
@@ -216,8 +221,8 @@ async function batch() {
         }
       }
     }
-    
-  } catch(e) {
+
+  } catch (e) {
     console.error(e)
   }
 }

start.sh

@@ -203,6 +203,7 @@ RUN_OPTIONS="
   --env \"T_JWT_ISSUER=$T_JWT_ISSUER\" \
   --env \"T_JWT_EXPIRES_IN=$T_JWT_EXPIRES_IN\" \
   --env \"T_ONLY_PROCESS_THESE_GROUPS=$T_ONLY_PROCESS_THESE_GROUPS\" \
+  --env \"T_LIMIT_NUMBER_OF_CHANGES=$T_LIMIT_NUMBER_OF_CHANGES\" \
   --volume $(pwd)/content-sets:/tangerine/content-sets:delegated \
   --volume $(pwd)/data/dat-output:/dat-output/ \
   --volume $(pwd)/data/reporting-worker-state.json:/reporting-worker-state.json \