Moved members-migrations out of the background jobs system

ghost/core/core/server/services/members/service.js

@@ -115,6 +115,66 @@ const initVerificationTrigger = () => {
   });
 };
 
+const membersMigrationJobName = 'members-migrations';
+
+/**
+ * Runs the historical Stripe backfills once per site, retried only after a
+ * failed run.
+ *
+ * The `jobs` row named `members-migrations` is the only guard: any status other
+ * than `failed` means the run was already attempted and is skipped forever. The
+ * row is written after the run, so a boot that dies mid-way retries next time.
+ * The row is written even when Stripe is not configured, matching the job-based
+ * version: a site that connects Stripe later never runs these 2021-era backfills.
+ * Two processes booting a site with no row will both run the backfills, where the
+ * job-based version claimed the row first. Accepted: every site that has booted
+ * since Ghost 5.6 has the row, and the backfills do nothing without Stripe.
+ *
+ * @TODO: Delete the backfills, this runner and the `jobs` rows in the next major
+ *
+ * @param {import('../stripe')} stripeService
+ */
+async function runStripeMigrations(stripeService) {
+  const existingJob = await models.Job.findOne({ name: membersMigrationJobName });
+
+  if (existingJob && existingJob.get('status') !== 'failed') {
+    logging.info(`Stripe ${membersMigrationJobName} skipped because it has already run`);
+    return;
+  }
+
+  const startedAt = Date.now();
+  logging.info(`Stripe ${membersMigrationJobName} started`);
+
+  let status = 'finished';
+  try {
+    await stripeService.migrations.execute();
+    logging.info(`Stripe ${membersMigrationJobName} completed in ${Date.now() - startedAt}ms`);
+  } catch (err) {
+    status = 'failed';
+    logging.error(
+      err,
+      `Stripe ${membersMigrationJobName} failed after ${Date.now() - startedAt}ms`,
+    );
+  }
+
+  const attrs = { status, started_at: new Date(startedAt), finished_at: new Date() };
+  if (existingJob) {
+    await models.Job.edit(attrs, { id: existingJob.id });
+    return;
+  }
+
+  try {
+    await models.Job.add({ name: membersMigrationJobName, ...attrs });
+  } catch (err) {
+    // Two processes booting a site with no row yet both get here, and the unique
+    // index on jobs.name rejects the second insert. The row exists, so move on.
+    if (!(await models.Job.findOne({ name: membersMigrationJobName }))) {
+      throw err;
+    }
+    logging.warn(`Stripe ${membersMigrationJobName} row was already written by another process`);
+  }
+}
+
 module.exports = {
   async init() {
     const stripeService = require('../stripe');
@@ -182,39 +242,7 @@ module.exports = {
       values: metafields.values,
     });
 
-    if (!env?.startsWith('testing')) {
-      const membersMigrationJobName = 'members-migrations';
-      if (!(await jobsService.hasExecutedSuccessfully(membersMigrationJobName))) {
-        logging.info(`[Background Job] ${membersMigrationJobName} queued`);
-        jobsService.addOneOffJob({
-          name: membersMigrationJobName,
-          offloaded: false,
-          job: async () => {
-            const startedAt = Date.now();
-            logging.info(`[Background Job] ${membersMigrationJobName} started`);
-            try {
-              const result = await stripeService.migrations.execute();
-              logging.info(
-                `[Background Job] ${membersMigrationJobName} completed in ${Date.now() - startedAt}ms`,
-              );
-              return result;
-            } catch (err) {
-              logging.error(
-                err,
-                `[Background Job] ${membersMigrationJobName} failed after ${Date.now() - startedAt}ms`,
-              );
-              throw err;
-            }
-          },
-        });
-
-        await jobsService.awaitOneOffCompletion(membersMigrationJobName);
-      } else {
-        logging.info(
-          `[Background Job] ${membersMigrationJobName} skipped because it has already run`,
-        );
-      }
-    }
+    await runStripeMigrations(stripeService);
   },
   contentGating: require('./content-gating'),
 

ghost/core/test/e2e-server/services/members-migrations.test.ts

@@ -0,0 +1,136 @@
+import assert from 'node:assert/strict';
+import sinon from 'sinon';
+
+const logging = require('@tryghost/logging');
+const { agentProvider } = require('../../utils/e2e-framework');
+const db = require('../../../core/server/data/db');
+const models = require('../../../core/server/models');
+const jobsService = require('../../../core/server/services/jobs');
+const membersService = require('../../../core/server/services/members');
+const stripeService = require('../../../core/server/services/stripe');
+
+const JOB_NAME = 'members-migrations';
+const STALE = new Date('2020-01-01T00:00:00Z');
+
+describe('Members migrations on boot', function () {
+  beforeAll(async function () {
+    await agentProvider.getAdminAPIAgent();
+  });
+
+  afterEach(function () {
+    sinon.restore();
+  });
+
+  it('records a finished job row on a fresh boot', async function () {
+    const job = await models.Job.findOne({ name: JOB_NAME });
+
+    assert.ok(job, 'expected a members-migrations row to be written during boot');
+    assert.equal(job.get('status'), 'finished');
+    assert.ok(job.get('started_at') instanceof Date);
+    assert.ok(job.get('finished_at') instanceof Date);
+  });
+
+  describe('when the members service initialises again', function () {
+    let jobId: string;
+
+    beforeEach(async function () {
+      const attrs = { status: 'finished', started_at: STALE, finished_at: STALE };
+      const job = await models.Job.findOne({ name: JOB_NAME });
+      jobId = job
+        ? (await models.Job.edit(attrs, { id: job.id })).id
+        : (await models.Job.add({ name: JOB_NAME, ...attrs })).id;
+    });
+
+    it('skips the migrations and writes nothing when the row already exists', async function () {
+      const execute = sinon.stub(stripeService.migrations, 'execute').resolves();
+      const add = sinon.spy(models.Job, 'add');
+      const edit = sinon.spy(models.Job, 'edit');
+
+      await membersService.init();
+
+      sinon.assert.notCalled(execute);
+      sinon.assert.notCalled(add);
+      sinon.assert.notCalled(edit);
+    });
+
+    it('re-runs the migrations and updates the row in place when the previous run failed', async function () {
+      await models.Job.edit({ status: 'failed' }, { id: jobId });
+      const execute = sinon.stub(stripeService.migrations, 'execute').resolves();
+      const add = sinon.spy(models.Job, 'add');
+      const addOneOffJob = sinon.spy(jobsService, 'addOneOffJob');
+      const awaitOneOffCompletion = sinon.spy(jobsService, 'awaitOneOffCompletion');
+
+      await membersService.init();
+
+      sinon.assert.calledOnce(execute);
+      sinon.assert.notCalled(add);
+      sinon.assert.notCalled(addOneOffJob);
+      sinon.assert.notCalled(awaitOneOffCompletion);
+
+      const after = await models.Job.findOne({ name: JOB_NAME });
+      assert.equal(after.id, jobId);
+      assert.equal(after.get('status'), 'finished');
+      assert.ok(after.get('started_at') > STALE, 'expected started_at to be rewritten');
+      assert.ok(after.get('finished_at') >= after.get('started_at'));
+    });
+
+    it('marks the row failed and keeps booting when the migrations throw', async function () {
+      await models.Job.edit({ status: 'failed' }, { id: jobId });
+      sinon.stub(stripeService.migrations, 'execute').rejects(new Error('stripe exploded'));
+      const loggingError = sinon.stub(logging, 'error');
+
+      await membersService.init();
+
+      sinon.assert.calledWithMatch(
+        loggingError,
+        sinon.match.has('message', 'stripe exploded'),
+        /members-migrations failed after \d+ms/,
+      );
+
+      const after = await models.Job.findOne({ name: JOB_NAME });
+      assert.equal(after.id, jobId);
+      assert.equal(after.get('status'), 'failed');
+      assert.ok(after.get('started_at') > STALE, 'expected the failed run to rewrite the row');
+      assert.ok(after.get('finished_at') >= after.get('started_at'));
+    });
+
+    it('keeps booting when another process writes the row first', async function () {
+      await db.knex('jobs').where({ name: JOB_NAME }).del();
+      sinon.stub(stripeService.migrations, 'execute').resolves();
+      const add: sinon.SinonStub = sinon.stub(models.Job, 'add').callsFake(async function (
+        this: unknown,
+        ...args: unknown[]
+      ) {
+        // The other process inserts the row first, then our own insert hits the
+        // unique index on jobs.name. The runner never inspects the error itself:
+        // any failed insert with a row present takes the warn path.
+        await add.wrappedMethod.apply(this, args);
+        return add.wrappedMethod.apply(this, args);
+      });
+      const loggingWarn = sinon.stub(logging, 'warn');
+
+      await membersService.init();
+
+      sinon.assert.calledOnce(add);
+      sinon.assert.calledWithMatch(loggingWarn, /row was already written by another process/);
+
+      const rows = await db.knex('jobs').where({ name: JOB_NAME });
+      assert.equal(rows.length, 1);
+      assert.equal(rows[0].status, 'finished');
+    });
+
+    it('fails boot when the insert fails and no other process wrote the row', async function () {
+      await db.knex('jobs').where({ name: JOB_NAME }).del();
+      const execute = sinon.stub(stripeService.migrations, 'execute').resolves();
+      sinon.stub(models.Job, 'add').rejects(new Error('insert exploded'));
+      const loggingWarn = sinon.stub(logging, 'warn');
+
+      await assert.rejects(membersService.init(), { message: 'insert exploded' });
+
+      sinon.assert.calledOnce(execute);
+      sinon.assert.notCalled(loggingWarn);
+      const rows = await db.knex('jobs').where({ name: JOB_NAME });
+      assert.equal(rows.length, 0);
+    });
+  });
+});

ghost/core/test/integration/services/member-welcome-emails.test.js

@@ -35,7 +35,7 @@ describe('Member Welcome Emails Integration', function () {
   beforeAll(async function () {
     await testUtils.setup('default')();
     membersService = require('../../../core/server/services/members');
-    membersService.init();
+    await membersService.init();
     defaultEmailDesignSettingId = await db
       .knex('email_design_settings')
       .where('slug', 'default-automated-email')