إظهار HN: Restate – عمليات سير عمل متينة ذات زمن وصول منخفض لـ JavaScript/Java، في Rust

يضمن التنفيذ الدائم تشغيل التعليمات البرمجية بشكل موثوق حتى النهاية، حتى في حالة وجود حالات فشل.

‍

• تتم إعادة محاولة حالات الفشل والأخطاء تلقائيًا (ما لم يتم تصنيفها على أنها أخطاء طرفية)
• يمكن للوظائف حفظ نتائج كتل التعليمات البرمجية والإجراءات مثل RPC في المجلة. لا يتم إعادة تنفيذ الخطوات المكتملة أثناء إعادة المحاولة، ولكن يتم إعادة تشغيلها من المجلة.
• يتم إنشاء سير العمل باستخدام التعليمات البرمجية العادية وتدفق التحكم، دون الحاجة إلى خطوط DSL مخصصة.
• تتيح فترات النوم المتينة للرمز الانتظار والتعليق لمدة تصل إلى أشهر

‍

export default restate.service({  name: "roleUpdate",  handlers: {    applyRoleUpdate: async (ctx, update)=> {      const { userId, role, permissions }=update;      const applied=await ctx.run("apply new role", ()=>        applyUserRole(userId, role)      );      if (!applied) {        return;      }      for (const permission of permissions) {        await ctx.run("apply permission", ()=>          applyPermission(userId, permission)        );      }    }  }});

@Servicepublic class RoleUpdateService {    @Handler    public void applyRoleUpdate(Context ctx, Update update) {        boolean success=ctx.run("apply new role", BOOLEAN,            () -> applyUserRole(update.getUserId(), update.getRole()));        if (!success) {            return;        }        for (String permission : update.getPermissions()) {            ctx.run("apply permission",                () -> applyPermission(update.getUserId(), permission));        }    }}

يمكنك الانضمام بشكل موثوق إلى التعليمات البرمجية المتزامنة والأحداث غير المتزامنة مثل خطافات الويب

• يتم الاحتفاظ بخطافات الويب/الأحداث في سجل Restate ويتم تسليمها بشكل موثوق إلى الخدمات
• تنضم الوعود/العقود الآجلة المستمرة بسهولة إلى مسارات التعليمات البرمجية المتزامنة وغير المتزامنة
• يضمن التنفيذ الدائم إكمالًا موثوقًا به، سواء جاءت خطافات الويب بعد ميلي ثانية أو أشهر، وتجنب إعادة تنفيذ الخطوات المكتملة.

const paymentSvc=restate.service({  name: "payments",  handlers: {    processPayment: async (ctx, request)=> {      const webhookPromise=ctx.awakeable();      const paymentIntent=await ctx.run("stripe call", ()=>        createPaymentIntent({          request,          metadata: { restate_callback_id: webhookPromise.id }        })      );      if (paymentIntent.status==="processing") {        // synchronous response inconclusive, await webhook response        const paymentIntentFromWebhook=await webhookPromise.promise;        return verifyPayment(paymentIntentFromWebhook);      } else {        return verifyPayment(paymentIntent);      }    },    processWebhook: async (ctx)=> {      const paymentIntent=verifyAndParseEvent(ctx.request());      const webhookPromiseId=paymentIntent.metadata.restate_callback_id;      ctx.resolveAwakeable(webhookPromiseId, paymentIntent);    }  }});

@Servicepublic class PaymentService {  @Handler  public void processPayment(Context ctx, PaymentRequest request) {    var webhookFuture=ctx.awakeable(SERDE);    var payment=ctx.run("Stripe call", SERDE, () -> submitPayment(            request, Map.of("restate_callback_id", webhookFuture.id())    ));    if (payment.getStatus().equals("processing")) {      // synchronous response inconclusive, await webhook response      var updatedPayment=webhookFuture.await();      verifyPayment(updatedPayment);    } else {      verifyPayment(payment);    }  }  @Handler  public void processWebhook(Context ctx) {    var paymentEvent=verifyAndParseEvent(ctx.request());    String callbackId=paymentEvent.getMetadata().get("restate_callback_id");    ctx.awakeableHandle(callbackId).resolve(SERDE, paymentEvent);  }}

يتم تنفيذ جميع الوظائف التي يتم استدعاؤها من خلال Restate بشكل دائم وغير متزامن.

• نشر الوظائف غير المتزامنة بدون خادم أو كحاويات أو عمليات.
• وظائف الاتصال بشكل متزامن أو غير متزامن أو متأخر. أعد إرفاقه وانتظر من أي مكان.
• أنشئ أنماطًا غير متزامنة مثل التوزيع الموسع والمروحة الداخلية وسلاسل المهام والمهام الفرعية ببساطة باستخدام استدعاءات الوظائف والعقود الآجلة/الوعود.
• استخدم الموقتات المستمرة لجدولة المهام في المستقبل.
• استخدم قوائم الانتظار الافتراضية الدقيقة (عبر الكائنات الافتراضية) لفرض ترتيب المهام الصارم والتزامن

// ------ service (=worker) ------const asyncTaskService=restate.service({    name: "taskWorker",    handlers: { processPayment }});// ------ client ------const rs=clients.connect({ url: process.env.RESTATE_URL });const taskWorker=rs.serviceSendClient({ name: "taskWorker" });// submit the payment task app.post('/charge/:paymentId', async (req, res)=> {    const taskHandle=await taskWorker.processPayment(        { request: req.params },        SendOpts.from({ idempotencyKey: req.params.paymentId })    );    res.json(taskHandle);});// await the payment taskapp.get('/status', async (req,res)=> {        const taskHandle=req.body.json();        const paymentResult=await restate.result(taskHandle);         res.join(paymentResult);});

// --- start payment task ---server.createContext("/charge", httpExchange -> {  PaymentRequest req=parsePaymentRequest(httpExchange);  SendResponse handle=AsyncTaskServiceClient      .fromIngress(RESTATE_URI)      .send()      .processPayment(req, idempotencyKey(req.getPaymentId()));  respondJson(httpExchange, handle);});//  --- connect to payment result ---server.createContext("/status", httpExchange -> {  String handle=parseToHandle(httpExchange);  String response=IngressClient.defaultClient(RESTATE_URI)      .invocationHandle(handle, STRING)      .attach();  respond(httpExchange, response);});

قم بمعالجة الأحداث (على سبيل المثال من Kafka) باستخدام وظائف متينة مثل معالجات الأحداث واحصل على عمليات إعادة المحاولة الدقيقة ودلالات سير العمل كتعليمات برمجية.

• لا توجد اشتراكات في قائمة الانتظار، ولا توجد إدارة إزاحة يدوية، أو قياس، أو موازنة
• انشر منطق معالجة الأحداث كوظائف بدون خادم على FaaS
• احتفظ بحالة لمرة واحدة بالضبط، أو قم بتأخير الأحداث، أو تشغيل عدة خطوات غير متزامنة أو استدعاءات واجهة برمجة التطبيقات (API).
• إن دلالات قائمة الانتظار لكل مفتاح في Restate لا تعني المزيد من تأثيرات الانتظار الرئيسية

const eventEnricher=restate.object({  name: "eventEnricher",  handlers: {    userEvent: async (ctx, event)=> {      // remember event, time box 100 ms to collect features      // before emitting result      ctx.set("user", event);      ctx.serviceSendClient(eventEnricher, { delay: 100 }).emit();    },    featureEvent: async (ctx, featureEvent)=> {      // merge feature into event      const userEvent=(await ctx.get("user")) ?? {};      (userEvent.features ??=[]).push(featureEvent);      ctx.set("user", userEvent)    },    emit: async (ctx)=> {      emit(ctx.key, await ctx.get("user"));      ctx.clearAll();    }  }})

@VirtualObjectpublic class EventEnricher {    static final StateKey USER=StateKey.of("user", of(User.class));    @Handler    public void userEvent(ObjectContext ctx, User event) {        ctx.set(USER, event);        // time box 100 ms to collect features before emitting result        EventEnricherClient.fromContext(ctx, ctx.key())            .send(ofMillis(100)).emit();    }    @Handler    public void featureEvent(ObjectContext ctx, Feature event) {        User user=ctx.get(USER).orElse(new User());        user.addFeature(event);        ctx.set(USER, user);    }    @Handler    public void emit(ObjectContext ctx) {        send(ctx.key(), ctx.get(USER));        ctx.clearAll();    }}

قم بإنشاء مسارات عمل ومعالجات للأحداث تتعامل بشكل موثوق مع الإشارات والأحداث والمدخلات البشرية الخارجية.

‍

• استخدم الوعود/العقود الآجلة الدائمة لصياغة الإشارات والشروط بشكل بديهي
• قم بإنشاء إشارات من RPCs أو webhooks أو أحداث Kafka
• يتم الاحتفاظ بالإشارات والأحداث من خلال إعادة الحالة، دون الحاجة إلى قائمة انتظار

export default workflow({  name: "verify",  handlers: {    run: async (ctx, { email })=> {      const secret=ctx.run("generate secret", ()=>        crypto.randomUUID()      );      await ctx.run("send email", ()=> sendEmail({ email, secret }));      const clickSecret=await ctx.promise("email.clicked");      return clickSecret==secret;    },    click: (ctx, { secret })=> {      ctx.promise("email.clicked").resolve(secret);    },  },});

@Workflowpublic class SecretVerifier {    static final DurablePromiseKey EMAIL_CLICKED=            DurablePromiseKey.of("email_clicked", JsonSerdes.STRING);    @Workflow    public boolean run(WorkflowContext ctx, Email email) {        String secret=ctx.random().nextUUID().toString();        ctx.run("send email",            () -> sendEmailWithLink(email, secret));        String clickSecret=ctx.promise(EMAIL_CLICKED).awaitable().await();        return clickSecret.equals(secret);    }    @Handler    public void click(SharedWorkflowContext ctx, String secret) {        ctx.promiseHandle(EMAIL_CLICKED).resolve(secret);    }}

أضف العجز إلى أي معالج RPC أو حدث.

‍

• يقبل كل استدعاء لمعالج الأحداث وRPC مفتاحًا غير فعال
• استخدم مفاتيح العجز لإعادة الارتباط باستدعاء مستمر
• تكون الاستدعاءات الواردة من سياق التنفيذ الدائم غير فعالة تلقائيًا

const rs=restate.connect({ url: process.env.RESTATE_URL });app.get('/reserve/:product/:reservationId', async (req, res)=> {  const { product, reservationId }=req.params;  const products=rs.serviceClient(ProductService);  const reservation=await products.reserve(    product,    Opts.from({ idempotencyKey : reservationId })  );  res.json(reservation);})

server.createContext("/reserve", httpExchange -> {      ReservationRequest req=parseRequest(httpExchange.getRequestBody());      // derive an idempotency key from the parameters      var idempotencyOps=CallRequestOptions.DEFAULT          .withIdempotency(req.getReservationId());      // add idempotency opts to the request to let the service automatically      // fuse repeated requests      Reservation reservation=ProductServiceClient          .fromIngress(RESTATE_RUNTIME_ENDPOINT)          .reserve(req.getProduct(), idempotencyOps);      sendResponse(httpExchange, reservation);    });

ينفذ قصصًا قوية وأنماط تعويض: معاملات طويلة الأمد تتراجع عن الإجراءات السابقة عندما تحتاج إلى الإجهاض والتراجع.

‍

• التقط بشكل موثوق بعد الفشل في الحصول على التعويضات
• ضمان حدوث التعويضات حتى في حالة الفشل أثناء مرحلة التعويض
• استخدم آليات الاستثناء/الخطأ القياسية وتحكم في التدفق بدلاً من خطوط DSL المعقدة.

async function reservation(ctx, products) {  const reservations=[];  try {    for (const product of products) {      const reservation=await ctx.run(`reserve ${product}`,          ()=> reserve(product));      reservations.push(reservation);    }  } catch (error) {    if (error instanceof TerminalError) {      for (const reservation of reservations) {        await ctx.run("undo reserve", ()=>             cancelReservation(reservation));      }    }    throw error;  }}

@Handlerpublic void reserveAllProducts(Context ctx, Product[] products) {    final List reservations=new ArrayList>();    try {        for (Product product : products) {            Reservation res=ctx.run("Reserve " + product.getId(),                RESERVE_SERDE, () -> reserve(product)            );            reservations.add(res);        }    } catch (TerminalException e) {        reservations.forEach(res -> {            ctx.run("Undo reservation", () -> cancelReservation(res));        });        throw e;    }}

قم بإنشاء أجهزة حالة متسقة وقابلة للتطوير بدون قواعد بيانات أو معاملات
‍

• قم بتشغيل الملايين من أجهزة الحالة التي تحافظ على الحالة مباشرة في سياق معالجاتها
• تلتزم تغييرات الحالة ذريًا بتنفيذ الوظيفة، لتحقيق الاتساق الصلب
• دلالات الكاتب الواحد لنموذج التزامن البسيط الميت. قائمة انتظار افتراضية لكل جهاز حالة لتحقيق الكفاءة وقابلية التوسع.
• يمكن أن يكون انتقال الحالة بمثابة سير عمل مع جميع الميزات من التنفيذ الدائم

const paymentSvc=restate.object({  name: "payments",  handlers: {    makePayment: async (ctx, payment)=> {      const paymentId=ctx.key;      switch (await ctx.get("status")) {        case "CANCELLED":            return `${paymentId} was cancelled before`;        case "SUCCESS":            return `${paymentId} previously completed`;      }      wireFunds(payment);      ctx.set("status", "SUCCESS");      ctx.set("payment", payment);    },    cancelPayment: async (ctx)=> {      const status=await ctx.get("status");      if (status==="SUCCESS") {        const payment=await ctx.get("payment");        refund(payment);      }      ctx.set("status", "CANCELLED");    }  }});

@VirtualObjectpublic class PaymentStateMachine {  @Handler  public String makePayment(ObjectContext ctx, PaymentRequest payment) {    String paymentId=ctx.key();    switch (ctx.get(STATE_STATUS).orElse(NEW)) {      case CANCELLED: return paymentId + " was cancelled before";      case SUCCESS:   return paymentId + " was previously completed";    }    wireFunds(payment);    ctx.set(STATE_STATUS, SUCCESS);    ctx.set(STATE_PAYMENT_REQUEST, payment);    return paymentId + " was successfully processed";  }  @Handler  public void cancelPayment(ObjectContext ctx) {    Status status=ctx.get(STATE_STATUS).orElse(NEW);    if (status==SUCCESS) {      PaymentRequest payment=ctx.get(STATE_PAYMENT_REQUEST).get();      refund(payment);    }    ctx.set(STATE_STATUS, CANCELLED);  }}