mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 16:09:05 +00:00
fix(license): order clicked refresh outcomes by the server's own attempt time
POST /api/license/refresh and /rotate now return attempt_at, the refresh.last_attempt_at this click recorded, read from the refresh status right after the click and kept only when it is newer than the pre-click attempt and carries this user and this outcome. Clicks refused before an attempt was recorded return none. The card keeps a clicked outcome until a strictly newer attempt arrives, falling back to the pre-click attempt when attempt_at is absent; the Date-header and browser-clock guesses are gone.
This commit is contained in:
@@ -68,6 +68,9 @@ struct RefreshBody {
|
||||
limits_refreshed: bool,
|
||||
#[serde(flatten)]
|
||||
report: RefreshReport,
|
||||
/// Unix microseconds of this click's own attempt, matching `refresh.last_attempt_at`.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
attempt_at: Option<i64>,
|
||||
}
|
||||
|
||||
#[derive(utoipa::ToSchema)]
|
||||
@@ -521,11 +524,13 @@ where
|
||||
fn refresh_response(
|
||||
limits_refreshed: bool,
|
||||
report: Result<RefreshReport, RefreshBusy>,
|
||||
attempt_at: Option<i64>,
|
||||
) -> Response {
|
||||
match report {
|
||||
Ok(report) => MetaHttpResponse::json(RefreshBody {
|
||||
limits_refreshed,
|
||||
report,
|
||||
attempt_at,
|
||||
}),
|
||||
Err(RefreshBusy) => (
|
||||
StatusCode::CONFLICT,
|
||||
@@ -540,6 +545,38 @@ fn refresh_response(
|
||||
}
|
||||
}
|
||||
|
||||
// the state is cluster-wide, so a later read may hold another node's attempt or an older one
|
||||
fn own_attempt_at(
|
||||
before: Option<i64>,
|
||||
after: &RefreshStatus,
|
||||
user: &str,
|
||||
report: &RefreshReport,
|
||||
) -> Option<i64> {
|
||||
let at = after
|
||||
.last_attempt_at
|
||||
.filter(|at| before.is_none_or(|before| *at > before))?;
|
||||
let same_outcome =
|
||||
serde_json::to_value(&after.last_outcome).ok() == serde_json::to_value(Some(report)).ok();
|
||||
(after.last_attempt_by.as_deref() == Some(user) && same_outcome).then_some(at)
|
||||
}
|
||||
|
||||
async fn attempt_status() -> RefreshStatus {
|
||||
refresh_status(true, true, false).await
|
||||
}
|
||||
|
||||
async fn click_response<Fut>(user: &str, limits_refreshed: bool, click: Fut) -> Response
|
||||
where
|
||||
Fut: std::future::Future<Output = Result<RefreshReport, RefreshBusy>>,
|
||||
{
|
||||
let before = attempt_status().await.last_attempt_at;
|
||||
let report = click.await;
|
||||
let attempt_at = match &report {
|
||||
Ok(report) => own_attempt_at(before, &attempt_status().await, user, report),
|
||||
Err(RefreshBusy) => None,
|
||||
};
|
||||
refresh_response(limits_refreshed, report, attempt_at)
|
||||
}
|
||||
|
||||
#[utoipa::path(
|
||||
get,
|
||||
path = "/license",
|
||||
@@ -721,10 +758,8 @@ pub async fn refresh_license_limits(Headers(user_email): Headers<UserEmail>) ->
|
||||
}
|
||||
|
||||
let limits_refreshed = get_license().await.is_some() && license::update().await.is_ok();
|
||||
refresh_response(
|
||||
limits_refreshed,
|
||||
click_refresh(&email, || async { license::update().await }).await,
|
||||
)
|
||||
let click = click_refresh(&email, || async { license::update().await });
|
||||
click_response(&email, limits_refreshed, click).await
|
||||
}
|
||||
|
||||
pub async fn rotate_license_key(Headers(user_email): Headers<UserEmail>) -> Response {
|
||||
@@ -734,7 +769,7 @@ pub async fn rotate_license_key(Headers(user_email): Headers<UserEmail>) -> Resp
|
||||
}
|
||||
|
||||
let limits_refreshed = get_license().await.is_some() && license::update().await.is_ok();
|
||||
refresh_response(limits_refreshed, rotate_instance_key(&email).await)
|
||||
click_response(&email, limits_refreshed, rotate_instance_key(&email)).await
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -1500,6 +1535,68 @@ mod tests {
|
||||
assert_eq!(body["error"], "no_license");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn refresh_returns_its_own_attempt_time_and_none_when_refused_before_one() {
|
||||
let _store = store_setup().await;
|
||||
load_refresh_license("online").await;
|
||||
// an unloaded stored key makes the click record a pending_reload attempt offline
|
||||
infra::db::get_db()
|
||||
.await
|
||||
.put(LICENSE_DB_KEY, "pending-other-key".into(), false, None)
|
||||
.await
|
||||
.unwrap();
|
||||
let (status, body) = refresh_as(STORE_ROOT).await;
|
||||
assert_eq!(status, StatusCode::OK, "{body}");
|
||||
assert_eq!(body["error"], "pending_reload");
|
||||
let attempt_at = body["attempt_at"].as_i64().expect("attempt_at");
|
||||
let (_, got) = get_as(STORE_ROOT).await;
|
||||
assert_eq!(got["refresh"]["last_attempt_at"].as_i64(), Some(attempt_at));
|
||||
|
||||
load_refresh_license("airgapped").await;
|
||||
let (status, body) = refresh_as(STORE_ROOT).await;
|
||||
assert_eq!(status, StatusCode::OK, "{body}");
|
||||
assert_eq!(body["error"], "airgapped");
|
||||
assert!(body.get("attempt_at").is_none(), "{body}");
|
||||
let (_, got) = get_as(STORE_ROOT).await;
|
||||
assert_eq!(got["refresh"]["last_attempt_at"].as_i64(), Some(attempt_at));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn own_attempt_at_takes_only_a_newer_attempt_matching_this_click() {
|
||||
let report = RefreshReport {
|
||||
outcome: Some(o2_enterprise::enterprise::license::RefreshOutcome::Unchanged),
|
||||
..Default::default()
|
||||
};
|
||||
let after = |at: Option<i64>, by: &str, outcome: Option<RefreshReport>| RefreshStatus {
|
||||
last_attempt_at: at,
|
||||
last_attempt_by: Some(by.into()),
|
||||
last_outcome: outcome,
|
||||
..refresh_status_fixture()
|
||||
};
|
||||
let mine = after(Some(20), "me@test", Some(report.clone()));
|
||||
assert_eq!(
|
||||
own_attempt_at(Some(10), &mine, "me@test", &report),
|
||||
Some(20)
|
||||
);
|
||||
assert_eq!(own_attempt_at(None, &mine, "me@test", &report), Some(20));
|
||||
assert_eq!(own_attempt_at(Some(20), &mine, "me@test", &report), None);
|
||||
assert_eq!(own_attempt_at(Some(30), &mine, "me@test", &report), None);
|
||||
let other_user = after(Some(20), "other@test", Some(report.clone()));
|
||||
assert_eq!(
|
||||
own_attempt_at(Some(10), &other_user, "me@test", &report),
|
||||
None
|
||||
);
|
||||
let other_outcome = after(Some(20), "me@test", Some(RefreshReport::default()));
|
||||
assert_eq!(
|
||||
own_attempt_at(Some(10), &other_outcome, "me@test", &report),
|
||||
None
|
||||
);
|
||||
assert_eq!(
|
||||
own_attempt_at(Some(10), &after(None, "me@test", None), "me@test", &report),
|
||||
None
|
||||
);
|
||||
}
|
||||
|
||||
fn refresh_status_fixture() -> o2_enterprise::enterprise::license::RefreshStatus {
|
||||
o2_enterprise::enterprise::license::RefreshStatus {
|
||||
mode: "online",
|
||||
|
||||
@@ -261,10 +261,11 @@ it.each([
|
||||
["rotate", "Installation key rotated."],
|
||||
])("a newer server attempt replaces the clicked %s outcome", async (action, clicked) => {
|
||||
const body = { outcome: "renewed", renewed: { license_id: "old", expires_at: 1825113600000000 } };
|
||||
refresh.mockResolvedValue({ ok: true, body });
|
||||
rotate.mockResolvedValue({ ok: true, body: { outcome: "updated" } });
|
||||
vi.useFakeTimers({ toFake: ["Date"] });
|
||||
vi.setSystemTime(1792748530000);
|
||||
refresh.mockResolvedValue({ ok: true, body: { ...body, attempt_at: 1792748520000000 } });
|
||||
rotate.mockResolvedValue({
|
||||
ok: true,
|
||||
body: { outcome: "updated", attempt_at: 1792748520000000 },
|
||||
});
|
||||
const w = card();
|
||||
await w.get(`[data-test="license-online-${action}"]`).trigger("click");
|
||||
await flushPromises();
|
||||
@@ -280,6 +281,13 @@ it.each([
|
||||
describe("clicked outcome versus newer server attempts", () => {
|
||||
const atSecond = (s: number) => s * 1_000_000;
|
||||
const outcomeText = (w: VueWrapper) => w.get('[data-test="license-online-outcome"]').text();
|
||||
const blocked = (last_attempt_at: number) =>
|
||||
status({
|
||||
last_attempt_at,
|
||||
last_outcome: { error: "blocked" },
|
||||
last_error: "blocked",
|
||||
registration: "blocked",
|
||||
});
|
||||
beforeEach(() => vi.useFakeTimers({ toFake: ["Date"] }));
|
||||
|
||||
it("an attempt seen while the click is in flight does not hide a later server failure", async () => {
|
||||
@@ -294,17 +302,10 @@ describe("clicked outcome versus newer server attempts", () => {
|
||||
last_outcome: { outcome: "unchanged" },
|
||||
}),
|
||||
});
|
||||
finish({ ok: true, body: { outcome: "unchanged" } });
|
||||
finish({ ok: true, body: { outcome: "unchanged", attempt_at: atSecond(1792748520) } });
|
||||
await flushPromises();
|
||||
expect(outcomeText(w)).toBe("License is up to date");
|
||||
await w.setProps({
|
||||
refresh: status({
|
||||
last_attempt_at: atSecond(1792748580),
|
||||
last_outcome: { error: "blocked" },
|
||||
last_error: "blocked",
|
||||
registration: "blocked",
|
||||
}),
|
||||
});
|
||||
await w.setProps({ refresh: blocked(atSecond(1792748580)) });
|
||||
expect(outcomeText(w)).toContain("blocked");
|
||||
});
|
||||
|
||||
@@ -324,8 +325,31 @@ describe("clicked outcome versus newer server attempts", () => {
|
||||
expect(outcomeText(w)).toBe("License is up to date");
|
||||
});
|
||||
|
||||
it("an answer without an attempt time gives way only to an attempt newer than the pre-click one", async () => {
|
||||
vi.setSystemTime(1792748530000);
|
||||
refresh.mockResolvedValue({ ok: true, body: { error: "airgapped" } });
|
||||
const w = card({ last_outcome: { outcome: "unchanged" } });
|
||||
await w.get('[data-test="license-online-refresh"]').trigger("click");
|
||||
await flushPromises();
|
||||
const clicked = outcomeText(w);
|
||||
expect(clicked).not.toBe("License is up to date");
|
||||
await w.setProps({
|
||||
refresh: status({
|
||||
last_attempt_at: atSecond(1792748460),
|
||||
last_outcome: { error: "blocked" },
|
||||
}),
|
||||
});
|
||||
expect(outcomeText(w)).toBe(clicked);
|
||||
await w.setProps({ refresh: blocked(atSecond(1792748460) + 1) });
|
||||
expect(outcomeText(w)).toContain("blocked");
|
||||
});
|
||||
|
||||
it("an older or equal attempt never replaces the clicked outcome", async () => {
|
||||
vi.setSystemTime(1792748530000);
|
||||
refresh.mockResolvedValue({
|
||||
ok: true,
|
||||
body: { outcome: "unchanged", attempt_at: atSecond(1792748520) },
|
||||
});
|
||||
const w = card({ last_outcome: { error: "blocked" } });
|
||||
await w.get('[data-test="license-online-refresh"]').trigger("click");
|
||||
await flushPromises();
|
||||
@@ -345,9 +369,12 @@ describe("clicked outcome versus newer server attempts", () => {
|
||||
expect(outcomeText(w)).toBe("License is up to date");
|
||||
});
|
||||
|
||||
it("another node's attempt recorded before the click's response keeps the clicked outcome", async () => {
|
||||
it("another node's attempt recorded before the click's own keeps the clicked outcome", async () => {
|
||||
vi.setSystemTime(1792748530000);
|
||||
rotate.mockResolvedValue({ ok: true, body: { outcome: "updated" } });
|
||||
rotate.mockResolvedValue({
|
||||
ok: true,
|
||||
body: { outcome: "updated", attempt_at: atSecond(1792748529) },
|
||||
});
|
||||
const w = card();
|
||||
await w.get('[data-test="license-online-rotate"]').trigger("click");
|
||||
await flushPromises();
|
||||
@@ -360,26 +387,61 @@ describe("clicked outcome versus newer server attempts", () => {
|
||||
expect(outcomeText(w)).toBe("Installation key rotated.");
|
||||
});
|
||||
|
||||
it("the server's response time wins over a skewed browser clock", async () => {
|
||||
vi.setSystemTime(1893456000000);
|
||||
it("adopts a later completed attempt within the same second as the click's own", async () => {
|
||||
vi.setSystemTime(1792748530900);
|
||||
rotate.mockResolvedValue({
|
||||
ok: true,
|
||||
body: { outcome: "updated", attempt_at: atSecond(1792748530) + 100_000 },
|
||||
});
|
||||
const w = card();
|
||||
await w.get('[data-test="license-online-rotate"]').trigger("click");
|
||||
await flushPromises();
|
||||
await w.setProps({
|
||||
refresh: status({
|
||||
last_attempt_at: atSecond(1792748530) + 100_000,
|
||||
last_outcome: { outcome: "updated" },
|
||||
}),
|
||||
});
|
||||
expect(outcomeText(w)).toBe("Installation key rotated.");
|
||||
await w.setProps({ refresh: blocked(atSecond(1792748530) + 700_000) });
|
||||
expect(outcomeText(w)).toContain("blocked");
|
||||
});
|
||||
|
||||
it("adopts subsequent server attempts whatever an ahead browser clock says", async () => {
|
||||
vi.setSystemTime(1792748830200);
|
||||
refresh.mockResolvedValue({
|
||||
ok: true,
|
||||
body: { outcome: "unchanged" },
|
||||
respondedAt: atSecond(1792748530) + 999_999,
|
||||
body: { outcome: "unchanged", attempt_at: atSecond(1792748530) + 100_000 },
|
||||
});
|
||||
const w = card();
|
||||
await w.get('[data-test="license-online-refresh"]').trigger("click");
|
||||
await flushPromises();
|
||||
await w.setProps({
|
||||
refresh: status({ last_attempt_at: atSecond(1792748530) + 500_000, last_outcome: null }),
|
||||
});
|
||||
expect(outcomeText(w)).toBe("License is up to date");
|
||||
await w.setProps({
|
||||
refresh: status({
|
||||
last_attempt_at: atSecond(1792748540),
|
||||
last_outcome: { error: "blocked" },
|
||||
last_attempt_at: atSecond(1792748530) + 100_000,
|
||||
last_outcome: { outcome: "unchanged" },
|
||||
}),
|
||||
});
|
||||
expect(outcomeText(w)).toBe("License is up to date");
|
||||
await w.setProps({ refresh: blocked(atSecond(1792748540) + 100_000) });
|
||||
expect(outcomeText(w)).toContain("blocked");
|
||||
});
|
||||
|
||||
it("keeps the rotation confirmation for its own attempt whatever a behind browser clock says", async () => {
|
||||
vi.setSystemTime(1792748470200);
|
||||
rotate.mockResolvedValue({
|
||||
ok: true,
|
||||
body: { outcome: "updated", attempt_at: atSecond(1792748530) + 100_000 },
|
||||
});
|
||||
const w = card();
|
||||
await w.get('[data-test="license-online-rotate"]').trigger("click");
|
||||
await flushPromises();
|
||||
await w.setProps({
|
||||
refresh: status({
|
||||
last_attempt_at: atSecond(1792748530) + 100_000,
|
||||
last_outcome: { outcome: "updated", waivers_added: 0, waivers_removed: 0 },
|
||||
}),
|
||||
});
|
||||
expect(outcomeText(w)).toBe("Installation key rotated.");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -319,8 +319,8 @@ const run = async (rotate: boolean) => {
|
||||
? rotateInstanceKey(() => emit("refreshed"))
|
||||
: refreshLicenseLimits(() => emit("refreshed")));
|
||||
result.value = clicked;
|
||||
// a refused click recorded no attempt; an answered one recorded its own by the time it responded
|
||||
clearAfter = clicked.ok ? Math.max(before, clicked.respondedAt ?? Date.now() * 1000) : before;
|
||||
// no attempt_at means no attempt was recorded or an older server answered
|
||||
clearAfter = (typeof clicked.body === "object" ? clicked.body.attempt_at : undefined) ?? before;
|
||||
dropClickedOutcome();
|
||||
emit("refreshed");
|
||||
} finally {
|
||||
|
||||
@@ -83,23 +83,17 @@ describe("useLicenseRefresh", () => {
|
||||
expect(fetchQuery).toHaveBeenCalledWith(expect.objectContaining({ queryKey: ["license"] }));
|
||||
});
|
||||
|
||||
it("reports the latest server time the response can carry from its Date header", async () => {
|
||||
it("passes the server's attempt time through and derives no time of its own", async () => {
|
||||
vi.mocked(licenseServer.refresh_license_limits).mockResolvedValue({
|
||||
status: 200,
|
||||
data: { outcome: "unchanged" },
|
||||
data: { outcome: "unchanged", attempt_at: 1792748530100000 },
|
||||
headers: { date: "Fri, 23 Oct 2026 03:02:00 GMT" },
|
||||
} as never);
|
||||
const ok = await useLicenseRefresh().refreshLicenseLimits();
|
||||
expect(ok.respondedAt).toBe(Date.parse("2026-10-23T03:02:00Z") * 1000 + 999_999);
|
||||
});
|
||||
|
||||
it("leaves the response time unset without a readable Date header", async () => {
|
||||
vi.mocked(licenseServer.refresh_license_limits).mockResolvedValue({
|
||||
expect(await useLicenseRefresh().refreshLicenseLimits()).toStrictEqual({
|
||||
ok: true,
|
||||
body: { outcome: "unchanged", attempt_at: 1792748530100000 },
|
||||
status: 200,
|
||||
data: {},
|
||||
headers: {},
|
||||
} as never);
|
||||
expect((await useLicenseRefresh().refreshLicenseLimits()).respondedAt).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
it("a 403 returns its status and schedules nothing", async () => {
|
||||
|
||||
@@ -22,12 +22,6 @@ import { refreshLicenseMutation, licenseQuery } from "@/services/license_server.
|
||||
import type { RefreshBody, RefreshResult } from "@/ts/interfaces/licenseRefresh";
|
||||
import { LICENSE_REFRESH_SETTLE_MS } from "@/utils/licenseWaivers";
|
||||
|
||||
// Date has whole seconds, so the response may have left up to 999_999 us after it
|
||||
const respondedAt = (date?: string) => {
|
||||
const ms = Date.parse(date ?? "");
|
||||
return Number.isNaN(ms) ? undefined : ms * 1000 + 999_999;
|
||||
};
|
||||
|
||||
export function useLicenseRefresh() {
|
||||
const store = useStore();
|
||||
const refreshMutation = useMutation(refreshLicenseMutation(), queryClient);
|
||||
@@ -82,12 +76,7 @@ export function useLicenseRefresh() {
|
||||
? rotateMutation.mutateAsync()
|
||||
: refreshMutation.mutateAsync());
|
||||
scheduleReload(onSettled);
|
||||
return {
|
||||
ok: true,
|
||||
body: response.data,
|
||||
status: response.status,
|
||||
respondedAt: respondedAt(response.headers?.date),
|
||||
};
|
||||
return { ok: true, body: response.data, status: response.status };
|
||||
} catch (error) {
|
||||
const e = error as { response?: { status?: number; data?: RefreshBody } };
|
||||
if (e.response?.status === 409) scheduleReload(onSettled);
|
||||
|
||||
@@ -20,6 +20,8 @@ export interface RefreshReport {
|
||||
export type RefreshBody =
|
||||
| (RefreshReport & {
|
||||
limits_refreshed?: boolean;
|
||||
/** Unix microseconds of this click's own attempt, matching `refresh.last_attempt_at`. */
|
||||
attempt_at?: number;
|
||||
code?: number;
|
||||
reason?: string;
|
||||
message?: string;
|
||||
@@ -50,6 +52,4 @@ export interface RefreshResult {
|
||||
ok: boolean;
|
||||
body?: RefreshBody;
|
||||
status?: number;
|
||||
/** Latest server time, in Unix microseconds, at which the response can have been sent. */
|
||||
respondedAt?: number;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user