Add explanatory comments for keys used in example (#296)

* Add explanatory comments for keys used in example

* Improve printing and cleanup

* Formatting

Author: gvdongen

go/patterns-use-cases/src/cron/cron.go

@@ -13,18 +13,18 @@ type JobRequest struct {
 	CronExpression string `json:"cronExpression"` // The cron expression e.g. "0 0 * * *" (every day at midnight)
 	Service        string `json:"service"`
 	Method         string `json:"method"`            // Handler to execute with this schedule
-	Key            string `json:"key,omitempty"`     // Optional: Virtual Object key to call
+	Key            string `json:"key,omitempty"`     // Optional: Virtual Object key of task to call
 	Payload        string `json:"payload,omitempty"` // Optional payload to pass to the handler
 }
 
 // JobInfo represents the stored job information
 type JobInfo struct {
-	Req               JobRequest `json:"req"`
+	Request           JobRequest `json:"request"`
 	NextExecutionTime time.Time  `json:"next_execution_time"`
 	NextExecutionID   string     `json:"next_execution_id"`
 }
 
-const JOB_KEY = "job" // Key for storing job information in the Restate object
+const JOB_STATE = "job-state" // K/V state key for storing job info in the Restate
 
 // CronJobInitiator service for creating new cron jobs
 //
@@ -44,9 +44,11 @@ const JOB_KEY = "job" // Key for storing job information in the Restate object
 type CronJobInitiator struct{}
 
 func (CronJobInitiator) Create(ctx restate.Context, req JobRequest) (string, error) {
+	// Create a new job ID and initiate the cron job object for that ID
+	// We can then address this job object by its ID
 	jobID := restate.Rand(ctx).UUID().String()
 
-	fmt.Printf("Creating new cron job with ID:", jobID, "for service:", req.Service, "method:", req.Method)
+	fmt.Printf("Creating new cron job with ID %s for service %s and method %s", jobID, req.Service, req.Method)
 	job, err := restate.Object[*JobInfo](ctx, "CronJob", jobID, "Initiate").Request(req)
 	if err != nil {
 		return "", err
@@ -59,59 +61,59 @@ func (CronJobInitiator) Create(ctx restate.Context, req JobRequest) (string, err
 type CronJob struct{}
 
 func (CronJob) Initiate(ctx restate.ObjectContext, req JobRequest) (*JobInfo, error) {
-	// Check if job already exists
-	job, err := restate.Get[*JobInfo](ctx, JOB_KEY)
+	// Check if jobState already exists
+	jobState, err := restate.Get[*JobInfo](ctx, JOB_STATE)
 	if err != nil {
 		return nil, err
 	}
-	if job != nil {
-		return nil, restate.TerminalErrorf("job already exists for this ID", 500)
+	if jobState != nil {
+		return nil, restate.TerminalErrorf("jobState already exists for this ID", 500)
 	}
 
 	return scheduleNextExecution(ctx, req)
 }
 
-func (CronJob) Execute(ctx restate.ObjectContext, req JobRequest) error {
+func (CronJob) Execute(ctx restate.ObjectContext) error {
 	// Get the job information
-	job, err := restate.Get[*JobInfo](ctx, JOB_KEY)
+	jobState, err := restate.Get[*JobInfo](ctx, JOB_STATE)
 	if err != nil {
 		return err
 	}
-	if job == nil {
+	if jobState == nil {
 		return restate.TerminalErrorf("job not found", 500)
 	}
 
 	// Add key if it's a virtual object call
-	fmt.Printf("Executing job with ID:", restate.Key(ctx), "for service:", req.Service, "method:", req.Method)
+	req := jobState.Request
+	fmt.Printf("Executing job with ID: %s for service %s for method %s", restate.Key(ctx), req.Service, req.Method)
 	if req.Key != "" {
 		restate.ObjectSend(ctx, req.Service, req.Key, req.Method).Send(req.Payload)
 	} else {
 		restate.ServiceSend(ctx, req.Service, req.Method).Send(req.Payload)
 	}
 
 	// Schedule the next execution
-	_, err = scheduleNextExecution(ctx, job.Req)
+	_, err = scheduleNextExecution(ctx, req)
 	return err
 }
 
 func (CronJob) Cancel(ctx restate.ObjectContext) error {
 	// Get the job to cancel the next execution
-	job, err := restate.Get[*JobInfo](ctx, JOB_KEY)
+	job, err := restate.Get[*JobInfo](ctx, JOB_STATE)
 	if err != nil {
 		return err
 	}
 	if job == nil {
 		return restate.TerminalErrorf("job not found for cancellation", 404)
 	}
 	restate.CancelInvocation(ctx, job.NextExecutionID)
-	restate.ObjectSend(ctx, "CronJob", restate.Key(ctx), "Cleanup").Send(nil)
 
 	restate.ClearAll(ctx)
 	return nil
 }
 
 func (CronJob) GetInfo(ctx restate.ObjectSharedContext) (*JobInfo, error) {
-	return restate.Get[*JobInfo](ctx, JOB_KEY)
+	return restate.Get[*JobInfo](ctx, JOB_STATE)
 }
 
 // scheduleNextExecution calculates and schedules the next execution of the cron job
@@ -132,15 +134,16 @@ func scheduleNextExecution(ctx restate.ObjectContext, req JobRequest) (*JobInfo,
 	nextTime := schedule.Next(currentTime)
 	delay := nextTime.Sub(currentTime)
 
-	// Schedule the next execution
-	handle := restate.ObjectSend(ctx, "CronJob", restate.Key(ctx), "Execute").Send(req, restate.WithDelay(delay))
+	// Schedule next execution for this job
+	thisJobID := restate.Key(ctx) // This got generated by the CronJobInitiator
+	handle := restate.ObjectSend(ctx, "CronJob", thisJobID, "Execute").Send(nil, restate.WithDelay(delay))
 
 	// Store the job information
-	job := &JobInfo{
-		Req:               req,
+	jobState := &JobInfo{
+		Request:           req,
 		NextExecutionTime: nextTime,
 		NextExecutionID:   handle.GetInvocationId(),
 	}
-	restate.Set(ctx, JOB_KEY, job)
-	return job, nil
+	restate.Set(ctx, JOB_STATE, jobState)
+	return jobState, nil
 }

go/patterns-use-cases/src/cron/task.go

@@ -10,7 +10,7 @@ import (
 type TaskService struct{}
 
 func (TaskService) ExecuteTask(ctx restate.Context, payload string) error {
-	fmt.Printf("Executing task: ", payload)
+	fmt.Printf("Executing task: %s", payload)
 	return nil
 }
 

java/patterns-use-cases/src/main/java/my/example/cron/Cron.java

@@ -37,18 +37,20 @@ public record JobRequest(
       String cronExpression, // e.g. "0 0 * * *" (every day at midnight)
       String service,
       String method, // Handler to execute with this schedule
-      Optional<String> key, // Optional Virtual Object key to call
+      Optional<String> key, // Optional Virtual Object key of the task to call
       Optional<String> payload) {} // Optional data to pass to the handler
 
-  public record JobInfo(JobRequest req, String nextExecutionTime, String nextExecutionId) {}
+  public record JobInfo(JobRequest request, String nextExecutionTime, String nextExecutionId) {}
 
   @Name("CronJobInitiator")
   @Service
   public static class JobInitiator {
     @Handler
-    public String create(Context ctx, JobRequest req) {
+    public String create(Context ctx, JobRequest request) {
+      // Create a new job ID and initiate the cron job object for that ID
+      // We can then address this job object by its ID
       var jobId = ctx.random().nextUUID().toString();
-      var cronJob = CronJobClient.fromContext(ctx, jobId).initiate(req).await();
+      var cronJob = CronJobClient.fromContext(ctx, jobId).initiate(request).await();
       return String.format(
           "Job created with ID %s and next execution time %s", jobId, cronJob.nextExecutionTime());
     }
@@ -58,37 +60,39 @@ public String create(Context ctx, JobRequest req) {
   @VirtualObject
   public static class Job {
 
-    private final StateKey<JobInfo> JOB = StateKey.of("job", JobInfo.class);
+    private final StateKey<JobInfo> JOB_STATE = StateKey.of("job-state", JobInfo.class);
     private final CronParser PARSER =
         new CronParser(CronDefinitionBuilder.instanceDefinitionFor(UNIX));
 
     @Handler
-    public JobInfo initiate(ObjectContext ctx, JobRequest req) {
-      if (ctx.get(JOB).isPresent()) {
+    public JobInfo initiate(ObjectContext ctx, JobRequest request) {
+      if (ctx.get(JOB_STATE).isPresent()) {
         throw new TerminalException("Job already exists for this ID");
       }
-      return scheduleNextExecution(ctx, req);
+      return scheduleNextExecution(ctx, request);
     }
 
     @Handler
     public void execute(ObjectContext ctx) {
-      JobRequest req = ctx.get(JOB).orElseThrow(() -> new TerminalException("Job not found")).req;
+      JobRequest request =
+          ctx.get(JOB_STATE).orElseThrow(() -> new TerminalException("Job not found")).request;
 
-      executeTask(ctx, req);
-      scheduleNextExecution(ctx, req);
+      executeTask(ctx, request);
+      scheduleNextExecution(ctx, request);
     }
 
     @Handler
     public void cancel(ObjectContext ctx) {
-      ctx.get(JOB).ifPresent(job -> ctx.invocationHandle(job.nextExecutionId).cancel());
+      ctx.get(JOB_STATE)
+          .ifPresent(jobState -> ctx.invocationHandle(jobState.nextExecutionId).cancel());
 
       // Clear the job state
       ctx.clearAll();
     }
 
     @Shared
     public Optional<JobInfo> getInfo(SharedObjectContext ctx) {
-      return ctx.get(JOB);
+      return ctx.get(JOB_STATE);
     }
 
     private void executeTask(ObjectContext ctx, JobRequest job) {
@@ -104,11 +108,11 @@ private void executeTask(ObjectContext ctx, JobRequest job) {
       ctx.send(request);
     }
 
-    private JobInfo scheduleNextExecution(ObjectContext ctx, JobRequest req) {
+    private JobInfo scheduleNextExecution(ObjectContext ctx, JobRequest request) {
       // Parse cron expression
       ExecutionTime executionTime;
       try {
-        executionTime = ExecutionTime.forCron(PARSER.parse(req.cronExpression));
+        executionTime = ExecutionTime.forCron(PARSER.parse(request.cronExpression));
       } catch (IllegalArgumentException e) {
         throw new TerminalException("Invalid cron expression: " + e.getMessage());
       }
@@ -124,13 +128,14 @@ private JobInfo scheduleNextExecution(ObjectContext ctx, JobRequest req) {
               .nextExecution(now)
               .orElseThrow(() -> new TerminalException("Cannot determine next execution time"));
 
-      // Schedule next execution
-      var handle = CronJobClient.fromContext(ctx, ctx.key()).send().execute(delay);
+      // Schedule next execution for this job
+      String thisJobId = ctx.key(); // This got generated by the CronJobInitiator
+      var handle = CronJobClient.fromContext(ctx, thisJobId).send().execute(delay);
 
       // Save job state
-      var job = new JobInfo(req, next.toString(), handle.invocationId());
-      ctx.set(JOB, job);
-      return job;
+      var jobState = new JobInfo(request, next.toString(), handle.invocationId());
+      ctx.set(JOB_STATE, jobState);
+      return jobState;
     }
   }
 }

typescript/patterns-use-cases/src/cron/cron_service.ts

@@ -6,17 +6,18 @@ type JobRequest = {
   cronExpression: string; // The cron expression e.g. "0 0 * * *" (every day at midnight)
   service: string;
   method: string; // Handler to execute with this schedule
-  key?: string; // Optional: Virtual Object key to call
+  key?: string; // Optional: Virtual Object key of task to call
   payload?: string; // Optional payload to pass to the handler
 };
 
 type JobInfo = {
-  req: JobRequest;
+  request: JobRequest;
   next_execution_time: string;
   next_execution_id: InvocationId;
 };
 
-const JOB = "job"; // Key for storing job information in the Restate object
+// Key of the K/V state that we store in Restate
+const JOB_STATE = "job-state";
 
 /*
  * A distributed cron service built with Restate that schedules tasks based on cron expressions.
@@ -35,9 +36,11 @@ const JOB = "job"; // Key for storing job information in the Restate object
 export const cronJobInitiator = restate.service({
   name: "CronJobInitiator",
   handlers: {
-    create: async (ctx: restate.Context, req: JobRequest) => {
+    create: async (ctx: restate.Context, request: JobRequest) => {
+      // Create a new job ID and initiate the cron job object for that ID
+      // We can then address this job object by its ID
       const jobId = ctx.rand.uuidv4();
-      const job = await ctx.objectClient(cronJob, jobId).initiate(req);
+      const job = await ctx.objectClient(cronJob, jobId).initiate(request);
       return `Job created with ID ${jobId} and next execution time ${job.next_execution_time}`;
     },
   },
@@ -46,21 +49,21 @@ export const cronJobInitiator = restate.service({
 export const cronJob = restate.object({
   name: "CronJob",
   handlers: {
-    initiate: async (ctx: restate.ObjectContext, req: JobRequest): Promise<JobInfo> => {
-      if (await ctx.get<JobInfo>(JOB)) {
+    initiate: async (ctx: restate.ObjectContext, request: JobRequest): Promise<JobInfo> => {
+      if (await ctx.get<JobInfo>(JOB_STATE)) {
         throw new TerminalError("Job already exists for this ID.");
       }
 
-      return await scheduleNextExecution(ctx, req);
+      return await scheduleNextExecution(ctx, request);
     },
     execute: async (ctx: restate.ObjectContext) => {
-      const job = await ctx.get<JobInfo>(JOB);
-      if (!job) {
+      const jobState = await ctx.get<JobInfo>(JOB_STATE);
+      if (!jobState) {
         throw new TerminalError("Job not found.");
       }
 
       // execute the task
-      const { service, method, key, payload } = job.req;
+      const { service, method, key, payload } = jobState.request;
       if (payload) {
         ctx.genericSend({
           service,
@@ -79,48 +82,49 @@ export const cronJob = restate.object({
         });
       }
 
-      await scheduleNextExecution(ctx, job.req);
+      await scheduleNextExecution(ctx, jobState.request);
     },
     cancel: async (ctx: restate.ObjectContext) => {
       // Cancel the next execution
-      const job = await ctx.get<JobInfo>(JOB);
-      if (job) {
-        ctx.cancel(job.next_execution_id);
+      const jobState = await ctx.get<JobInfo>(JOB_STATE);
+      if (jobState) {
+        ctx.cancel(jobState.next_execution_id);
       }
 
       // Clear the job state
       ctx.clearAll();
     },
-    getInfo: async (ctx: restate.ObjectSharedContext) => ctx.get<JobInfo>(JOB),
+    getInfo: async (ctx: restate.ObjectSharedContext) => ctx.get<JobInfo>(JOB_STATE),
   },
 });
 
 const scheduleNextExecution = async (
   ctx: restate.ObjectContext,
-  req: JobRequest
+  request: JobRequest
 ): Promise<JobInfo> => {
   // Parse cron expression
   // Persist current date in Restate for deterministic replay
   const currentDate = await ctx.date.now();
   let interval;
   try {
-    interval = CronExpressionParser.parse(req.cronExpression, { currentDate });
+    interval = CronExpressionParser.parse(request.cronExpression, { currentDate });
   } catch (e) {
     throw new TerminalError(`Invalid cron expression: ${(e as Error).message}`);
   }
 
   const next = interval.next().toDate();
   const delay = next.getTime() - currentDate;
 
-  // Schedule next execution
-  const handle = ctx.objectSendClient(cronJob, ctx.key, { delay }).execute();
+  // Schedule next execution for this job
+  const thisJobId = ctx.key; // This got generated by the CronJobInitiator
+  const handle = ctx.objectSendClient(cronJob, thisJobId, { delay }).execute();
 
   // Store the job information
-  const job = {
-    req,
+  const jobState = {
+    request,
     next_execution_time: next.toString(),
     next_execution_id: await handle.invocationId,
   };
-  ctx.set<JobInfo>(JOB, job);
-  return job;
+  ctx.set<JobInfo>(JOB_STATE, jobState);
+  return jobState;
 };