openvibers/openvibe.media Commits
Diff summary (exactly the two prescribed changes, nothing else) (#6)
Changes
@@@ -124,9 +124,11 @@ function isHealthy(provider, { purpose = 'playback' } = {}) {return providers.eligibleClasses(provider).length > 0;}--/** Measured 5-minute EWMA (ms), or null when the provider has no samples. */++/** Measured 5-minute EWMA (ms) of a network round-trip op (head, then get), or null when the++ * provider has no such samples. Presign is excluded: it is a local SigV4 operation, so its++ * EWMA is host CPU noise, not a network signal, and must not rank providers. */function measuredMs(provider) {-- const ms = signals.latencyEWMA(provider, 'presign') || signals.latencyEWMA(provider);++ const ms = signals.latencyEWMA(provider, 'head') || signals.latencyEWMA(provider, 'get');return ms > 0 ? ms : null;}@@@ -183,8 +183,9 @@ const { spawnSync } = require('child_process');config.objects.publicMaxMb = 500;config.objects.publicTargetMb = 256;-- // A retry resumes after the parts its checkpoint names.-- const resumed = await queue.enqueue({ appId: 'live', type: 'object.split', objectId: vodObj, params: { parts: 3 }, createdBy: 'test' });++ // A retry resumes after the parts its checkpoint names. `runAfterS` keeps the worker++ // (50 ms poll) from claiming the job before the checkpoint below is saved.++ const resumed = await queue.enqueue({ appId: 'live', type: 'object.split', objectId: vodObj, params: { parts: 3 }, createdBy: 'test', runAfterS: 1 });await queue.saveCheckpoint(resumed.job.id, { parts: [{ part: 1, object_id: 'med_done_before', start_seconds: 0, duration_seconds: 2, size_bytes: 10 }] });assert.ok(await waitFor(async () => (await queue.get(resumed.job.id)).status === 'succeeded'));const rr = queue.jobPublic(await queue.get(resumed.job.id)).result;@@@ -86,6 +86,10 @@ const crypto = require('crypto');await storage.tierConfig.init(storage.DEFAULTS);await require('../server/objects/tier-policy').init();const { runDrill, providerOfUrl } = require('../scripts/r2-eviction-drill');++ // Both buckets live on one local endpoint, so their measured latencies are noise: pin them equal so the++ // router ranks the copies by tier alone (a few ms of jitter could otherwise promote B2 over R2).++ const signals = require('../server/placement/signals');++ signals.latencyEWMA = () => 50;const conn = db.getDb();// Media serving /v/:id, as production does.
@@ -124,9 +124,11 @@ function isHealthy(provider, { purpose = 'playback' } = {}) {
@@ -124,9 +124,11 @@ function isHealthy(provider, { purpose = 'playback' } = {}) {
return providers.eligibleClasses(provider).length > 0;
return providers.eligibleClasses(provider).length > 0;
}
}
-/** Measured 5-minute EWMA (ms), or null when the provider has no samples. */
+/** Measured 5-minute EWMA (ms) of a network round-trip op (head, then get), or null when the
+ * provider has no such samples. Presign is excluded: it is a local SigV4 operation, so its
+ * EWMA is host CPU noise, not a network signal, and must not rank providers. */
function measuredMs(provider) {
function measuredMs(provider) {
- const ms = signals.latencyEWMA(provider, 'presign') || signals.latencyEWMA(provider);
+ const ms = signals.latencyEWMA(provider, 'head') || signals.latencyEWMA(provider, 'get');
return ms > 0 ? ms : null;
return ms > 0 ? ms : null;
}
}
@@ -183,8 +183,9 @@ const { spawnSync } = require('child_process');
@@ -183,8 +183,9 @@ const { spawnSync } = require('child_process');
config.objects.publicMaxMb = 500;
config.objects.publicMaxMb = 500;
config.objects.publicTargetMb = 256;
config.objects.publicTargetMb = 256;
- // A retry resumes after the parts its checkpoint names.
- const resumed = await queue.enqueue({ appId: 'live', type: 'object.split', objectId: vodObj, params: { parts: 3 }, createdBy: 'test' });
+ // A retry resumes after the parts its checkpoint names. `runAfterS` keeps the worker
+ // (50 ms poll) from claiming the job before the checkpoint below is saved.
+ const resumed = await queue.enqueue({ appId: 'live', type: 'object.split', objectId: vodObj, params: { parts: 3 }, createdBy: 'test', runAfterS: 1 });
await queue.saveCheckpoint(resumed.job.id, { parts: [{ part: 1, object_id: 'med_done_before', start_seconds: 0, duration_seconds: 2, size_bytes: 10 }] });
await queue.saveCheckpoint(resumed.job.id, { parts: [{ part: 1, object_id: 'med_done_before', start_seconds: 0, duration_seconds: 2, size_bytes: 10 }] });
assert.ok(await waitFor(async () => (await queue.get(resumed.job.id)).status === 'succeeded'));
assert.ok(await waitFor(async () => (await queue.get(resumed.job.id)).status === 'succeeded'));
const rr = queue.jobPublic(await queue.get(resumed.job.id)).result;
const rr = queue.jobPublic(await queue.get(resumed.job.id)).result;
@@ -86,6 +86,10 @@ const crypto = require('crypto');
@@ -86,6 +86,10 @@ const crypto = require('crypto');
await storage.tierConfig.init(storage.DEFAULTS);
await storage.tierConfig.init(storage.DEFAULTS);
await require('../server/objects/tier-policy').init();
await require('../server/objects/tier-policy').init();
const { runDrill, providerOfUrl } = require('../scripts/r2-eviction-drill');
const { runDrill, providerOfUrl } = require('../scripts/r2-eviction-drill');
+ // Both buckets live on one local endpoint, so their measured latencies are noise: pin them equal so the
+ // router ranks the copies by tier alone (a few ms of jitter could otherwise promote B2 over R2).
+ const signals = require('../server/placement/signals');
+ signals.latencyEWMA = () => 50;
const conn = db.getDb();
const conn = db.getDb();
// Media serving /v/:id, as production does.
// Media serving /v/:id, as production does.